如何判断Oracle Streams AQ入队的payload/消息大小是否超过32K

背景说明 #

根据Oracle Streams Advanced Queuing用户指南与参考文档说明："存储RAW类型的payload时，Oracle Streams AQ会创建带LOB列的队列表作为payload存储库。payload的最大大小由访问Oracle Streams AQ的编程接口决定：PL/SQL、Java和预编译器的限制为32K；OCI接口的限制为4G。"

核心问题 #

如何判断入队的payload/消息大小超过PL/SQL接口的32K限制？

初始存储过程代码 #

CREATE OR REPLACE procedure PRC_ordercreated(P_MSG in clob, P_MSGID out raw)
is
  V_ENQUEUEOPTIONS SYS.DBMS_AQ.ENQUEUE_OPTIONS_T;
  V_MESSAGEPROPERTIES SYS.DBMS_AQ.MESSAGE_PROPERTIES_T;

  V_QUEUENAME varchar2(35) := 'QUE_ordercreated';
begin
  V_MESSAGEPROPERTIES.USER_PROPERTY := SYS.ANYDATA.CONVERTTIMESTAMPTZ(systimestamp);

  /*  当payload消息超过32K时，消息将被存储在单独的表中
  */
  
  SYS.DBMS_AQ.ENQUEUE(
    QUEUE_NAME => V_QUEUENAME,
    PAYLOAD => SYS.UTL_RAW.CAST_TO_RAW(P_MSG),
    ENQUEUE_OPTIONS  => V_ENQUEUEOPTIONS,
    MESSAGE_PROPERTIES => V_MESSAGEPROPERTIES,
    MSGID => P_MSGID);
    
  insert into QUEUE_OVERSIZEDMESSAGE(
    MSGID,
    LARGEMESSAGE)
  values (
    P_MSGID,
    P_MSG);
end;
/

最终优化实现方案 #

通过SYS.UTL_RAW.LENGTH计算转换为RAW类型后的payload长度，和32000阈值做判断，超过阈值的大消息单独存入扩展表，AQ队列仅存储标记：

CREATE OR REPLACE procedure PRC_ENQUEUE(P_MSG in clob, P_MSGID out raw)
is
  V_ENQUEUEOPTIONS SYS.DBMS_AQ.ENQUEUE_OPTIONS_T;
  V_MESSAGEPROPERTIES SYS.DBMS_AQ.MESSAGE_PROPERTIES_T;
  V_QUEUENAME varchar2(16) := 'QUE_ORDERCREATED';
  V_MAXPAYLOADSIZE number := 32000;
begin
  V_MESSAGEPROPERTIES.USER_PROPERTY := SYS.ANYDATA.CONVERTTIMESTAMPTZ(systimestamp);

  /* 当payload消息超过32K时，消息将被存储在单独的表中
  */
  if SYS.UTL_RAW.LENGTH(SYS.UTL_RAW.CAST_TO_RAW(P_MSG)) > V_MAXPAYLOADSIZE then
    SYS.DBMS_AQ.ENQUEUE(
      QUEUE_NAME => V_QUEUENAME,
      PAYLOAD => SYS.UTL_RAW.CAST_TO_RAW('IsLargeMessage'),
      ENQUEUE_OPTIONS  => V_ENQUEUEOPTIONS,
      MESSAGE_PROPERTIES => V_MESSAGEPROPERTIES,
      MSGID => P_MSGID);

    insert into QUEUE_LARGEMESSAGE(
      MSGID,
      LARGEMESSAGE,
      CREATIONDATETIME,
      LASTMODIFICATIONDATETIME)
    values (
      P_MSGID,
      P_MSG,
      systimestamp,
      systimestamp);
  else
    SYS.DBMS_AQ.ENQUEUE(
      QUEUE_NAME => V_QUEUENAME,
      PAYLOAD => SYS.UTL_RAW.CAST_TO_RAW(P_MSG),
      ENQUEUE_OPTIONS  => V_ENQUEUEOPTIONS,
      MESSAGE_PROPERTIES => V_MESSAGEPROPERTIES,
      MSGID => P_MSGID);
  end if;
end;
/

内容的提问来源于stack exchange，提问作者PurpleGreen