在 PL/SQL 中创建队列订阅者的语法是什么？

Question

我正在尝试创建一个队列和一个在消息排队时触发的回调，但我无法触发回调。我做错了什么？

我有一个将消息入队的触发器，我可以在队列消息表上看到它，我可以手动将其出队并处理它，我只是无法让回调在入队时触发。

BEGIN    
DBMS_AQADM.CREATE_QUEUE_TABLE (
  queue_table        => 'queue_message_table',
  queue_payload_type => 'queue_message_type',
  multiple_consumers => TRUE);

DBMS_AQADM.CREATE_QUEUE (
  queue_name  => 'message_queue',
  queue_table => 'queue_message_table');
DBMS_AQADM.START_QUEUE (queue_name => 'message_queue');
END;

CREATE OR REPLACE PROCEDURE queue_callback(
  context RAW, reginfo  SYS.AQ$_REG_INFO, descr SYS.AQ$_DESCRIPTOR, payload  RAW, payloadl NUMBER) AS

    queue_options       DBMS_AQ.DEQUEUE_OPTIONS_T;
    message_properties  DBMS_AQ.MESSAGE_PROPERTIES_T;
    my_message          queue_message_type;
    ret                 varchar2(200);
    message_id          RAW(16);
BEGIN
    DBMS_OUTPUT.PUT_LINE('Callback');
    queue_options.msgid := descr.msg_id;
    queue_options.consumer_name := descr.consumer_name;

    DBMS_AQ.DEQUEUE(
        queue_name => descr.queue_name,
        dequeue_options => queue_options,
        message_properties => message_properties,
        payload => my_message,
        msgid => message_id );
    ret := handle_message(my_message);
    commit;
END;

BEGIN
  DBMS_AQADM.ADD_SUBSCRIBER (queue_name => 'message_queue',
    subscriber => SYS.AQ$_AGENT('queue_subscriber', 'message_queue',NULL));
  DBMS_AQ.REGISTER (
    SYS.AQ$_REG_INFO_LIST(
      SYS.AQ$_REG_INFO(
        'MESSAGE_QUEUE:QUEUE_SUBSCRIBER',
        DBMS_AQ.NAMESPACE_AQ,
        'plsql://QUEUE_CALLBACK',
        HEXTORAW('FF')
      )
    ), 1
  );
END;

Answer 1

乍一看，您似乎既没有开始排队 (dbms_aqadm.start_queue)，也没有将任何东西排入队列 (dbms_aq.enqueue)。

我建议关注这个demo。

Answer 2

您需要注意数据库版本。已经报告了一些关于 issues with Oracle Aq 的错误。 我特别关注this link 构建了我自己的示例，在 Oracle 11gR2 企业数据库中执行演示。我能够入队、出队、清除队列，但使用 Dbms_Aq.Register 创建的侦听器不起作用。 我运行了相同的示例，下载了一个 Oracle 11g R2 xe 数据库，它运行正常。

同样的示例在 Oracle 10gR2 实例中运行，并且运行良好。

在使用 aq 时需要注意一些事项：

• 使用适当的参数添加订阅者
• 使用适当的命名空间向 Dbms_Aq.Register 注册侦听器
• 使用多个消费者标志来声明队列表
• 对包使用适当的权限并处理队列
• 如果不起作用，在某些情况下使用队列的限定名称。

' 首先创建架构

connect / as sysdba
-- @?/rdbms/admin/dbmsaqad.sql --(install if you don't have aq installed yet)

-- create the user and permissions
create user aqadmin identified by aqadmin default tablespace users temporary tablespace temp;
GRANT create session TO aqadmin;
grant connect, resource to aqadmin;
GRANT aq_administrator_role TO aqadmin IDENTIFIED BY aqadmin;
GRANT execute ON dbms_aq TO aqadmin;
GRANT execute ON dbms_aqadm TO aqadmin;

创建 ddl 对象

CREATE TABLE demo_queue_message_table
( message VARCHAR2(4000) );

创建 aq 对象

create or replace type demo_queue_payload_type as object(message varchar2(4000)) ;
/

begin
  DBMS_AQADM.CREATE_QUEUE_TABLE (queue_table => 'demo_queue_table', queue_payload_type => 'demo_queue_payload_type',multiple_consumers => TRUE);
  DBMS_AQADM.CREATE_QUEUE (queue_name => 'demo_queue', queue_table => 'demo_queue_table');
  DBMS_AQADM.START_QUEUE('demo_queue');
end;
/

CREATE or replace PROCEDURE demo_queue_callback_procedure(
                 context  RAW,
                 reginfo  SYS.AQ$_REG_INFO,
                 descr    SYS.AQ$_DESCRIPTOR,
                 payload  RAW,
                 payloadl NUMBER
                 ) AS

   r_dequeue_options    DBMS_AQ.DEQUEUE_OPTIONS_T;
   r_message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
   v_message_handle     RAW(16);
   o_payload            demo_queue_payload_type;

BEGIN

   r_dequeue_options.msgid := descr.msg_id;
   r_dequeue_options.consumer_name := descr.consumer_name;

   DBMS_AQ.DEQUEUE(
      queue_name         => descr.queue_name,
      dequeue_options    => r_dequeue_options,
      message_properties => r_message_properties,
      payload            => o_payload,
      msgid              => v_message_handle
      );

   INSERT INTO demo_queue_message_table ( message )
   VALUES ( 'Message [' || o_payload.message || '] ' ||
            'dequeued at [' || TO_CHAR( SYSTIMESTAMP,
                                        'DD-MON-YYYY HH24:MI:SS.FF3' ) || ']' );
   COMMIT;

END;
/


BEGIN

   DBMS_AQADM.ADD_SUBSCRIBER (
      queue_name => 'demo_queue',
      subscriber => SYS.AQ$_AGENT(
                       'demo_queue_subscriber',
                       NULL,
                       NULL )
      );

    DBMS_AQ.REGISTER (
       SYS.AQ$_REG_INFO_LIST(
          SYS.AQ$_REG_INFO(
             'DEMO_QUEUE:DEMO_QUEUE_SUBSCRIBER',
             DBMS_AQ.NAMESPACE_AQ,
             'plsql://DEMO_QUEUE_CALLBACK_PROCEDURE',
             HEXTORAW('FF')
             )
          ),
       1
       );
END;
/

最后测试队列

DECLARE

   r_enqueue_options    DBMS_AQ.ENQUEUE_OPTIONS_T;
   r_message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
   v_message_handle     RAW(16);
   o_payload            demo_queue_payload_type;

BEGIN

   o_payload := demo_queue_payload_type(
                   TO_CHAR(SYSTIMESTAMP, 'DD-MON-YYYY HH24:MI:SS.FF3' )
                   );

   DBMS_AQ.ENQUEUE(
      queue_name         => 'demo_queue',
      enqueue_options    => r_enqueue_options,
      message_properties => r_message_properties,
      payload            => o_payload,
      msgid              => v_message_handle
      );

  COMMIT;

END;
/