Oracle Advanced Queuing 之04(优先级)

1.創建隊列及隊列表且启动之 aq_admin login

begin
--Create a table for queues 
dbms_aqadm.create_queue_table(queue_table=>'aq_msg_qtab3',queue_payload_type=>'AQ_ADMIN.AQ_MSG_TYPE'
,sort_list => 'priority,enq_time' --按优先级和入列时间排序
,multiple_consumers => true  --多消费者
,comment => 'queue for aq_msg_queue3'
);

  --Create a test queue
  dbms_aqadm.create_queue(
   queue_name  => 'aq_msg_queue3'
  ,queue_table => 'aq_msg_qtab3'
  ,queue_type => sys.dbms_aqadm.normal_queue
  ,max_retries => 3 --dequeue失败后重试次数
  ,retry_delay => 1 --重试前等待
  ,retention_time => 0 --dequeue后保持时间，不保持
);
                           
  --Start the queue                           
  dbms_aqadm.start_queue (queue_name => 'aq_msg_queue3');      
end;
/

begin
 dbms_aqadm.grant_queue_privilege (privilege => 'ALL',queue_name => 'AQ_ADMIN.AQ_MSG_QUEUE3',grantee => 'AQ_USER',grant_option => FALSE);
end;

2.创建消息订阅者 aq_admin login

begin  
  -- Configure the demo subscriber.
  dbms_aqadm.add_subscriber (queue_name => 'aq_msg_queue3',subscriber => sys.aq$_agent(name => 'subscriber_xag3',address  => null,protocol => 0));

end;
/

3.入队 aq_user login

declare
  v_queue_name varchar2(50);
  v_enqueue_options     dbms_aq.enqueue_options_t;
  v_message_properties  dbms_aq.message_properties_t;
  v_message_handle      raw(16);
  v_aq_msg              aq_admin.aq_msg_type;
begin
  v_queue_name:='aq_admin.aq_msg_queue3';
  v_aq_msg := aq_admin.aq_msg_type (11,'-','-','p1','content priority 1',sysdate);
  v_message_properties.priority := 1; --该消息的优先级别
  v_enqueue_options.visibility :=dbms_aq.immediate;
  dbms_aq.enqueue(
      queue_name => v_queue_name
      ,enqueue_options => v_enqueue_options
      ,message_properties => v_message_properties
      ,payload => v_aq_msg
      ,msgid => v_message_handle);
  dbms_output.put_line('encode success,msgid is ' || v_message_handle);
  commit;
end;

--改变 priority 的值为3，2，再入列两次，现在入列的顺序为1 3 2，
--我们希望的出列顺序为 1，2，3
 入队
 v_aq_msg := aq_admin.aq_msg_type (13,'-','-','p3','content priority 3',sysdate);
 v_message_properties.priority := 3; --该消息的优先级别
再入队
 v_aq_msg := aq_admin.aq_msg_type (12,'-','-','p2','content priority 2',sysdate);
 v_message_properties.priority := 2; --该消息的优先级别

select count(1) from aq_admin.aq_msg_qtab3;
-------------------------------------------------------
3

4.出队列(aq_user)

declare
 v_queue_name varchar2(50);
 v_aq_msg             aq_admin.aq_msg_type;
 v_message_handle RAW(16);
 v_dequeue_options DBMS_AQ.DEQUEUE_OPTIONS_T;
 v_message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
begin
 v_dequeue_options.visibility :=DBMS_AQ.IMMEDIATE;
 v_dequeue_options.consumer_name := 'subscriber_xag3';
 v_queue_name:='aq_admin.aq_msg_queue3';

 dbms_aq.dequeue(queue_name => v_queue_name,dequeue_options => v_dequeue_options,message_properties => v_message_properties,payload => v_aq_msg,msgid => v_message_handle);
 dbms_output.put_line('--------------------------------------- ');
 dbms_output.put_line('msg_seq :'         || v_aq_msg.msg_seq);
 dbms_output.put_line('msg_sender :'      || v_aq_msg.msg_sender);
 dbms_output.put_line('msg_receiver :'    || v_aq_msg.msg_receiver);
 dbms_output.put_line('msg_level :'       || v_aq_msg.msg_level);
 dbms_output.put_line('msg_content :'     || v_aq_msg.msg_content);
 dbms_output.put_line('msg_create_time :' || to_char(v_aq_msg.msg_create_time,'yyyy-mm-dd hh24:mi:ss'));
 dbms_output.put_line('current_time :'    || to_char(v_aq_msg.msg_create_time,'yyyy-mm-dd hh24:mi:ss'));
 commit;
end;
---------------------------------------------------------------------------------
以上出队脚本运行3次，依次得到如下结果：
msg_seq :11
msg_sender :-
msg_receiver :-
msg_level :p1
msg_content :content priority 1
msg_create_time :2020-11-13 22:37:08
current_time :2020-11-13 22:37:08
--------------------------------------- 
msg_seq :12
msg_sender :-
msg_receiver :-
msg_level :p2
msg_content :content priority 2
msg_create_time :2020-11-13 22:37:34
current_time :2020-11-13 22:37:34
--------------------------------------- 
msg_seq :13
msg_sender :-
msg_receiver :-
msg_level :p3
msg_content :content priority 3
msg_create_time :2020-11-13 22:37:20
current_time :2020-11-13 22:37:20

5.注册回调 aq_admin login

begin                                                          
  -- Register the procedure for dequeuing the messages received.
  dbms_aq.register(
    sys.aq$_reg_info_list(
      sys.aq$_reg_info('AQ_ADMIN.AQ_MSG_QUEUE3:SUBSCRIBER_XAG3',dbms_aq.namespace_aq, 
                       'plsql://AQ_USER.CALL_BACK_PCK.AQ_MSG_QUEUE1',hextoraw('FF'))
                      ),
      1);
end;

6.入队测试（aq_user)

declare
  v_queue_name varchar2(50);
  v_enqueue_options     dbms_aq.enqueue_options_t;
  v_message_properties  dbms_aq.message_properties_t;
  v_message_handle      raw(16);
  v_aq_msg              aq_admin.aq_msg_type;
begin
  v_queue_name:='aq_admin.aq_msg_queue3';
  v_aq_msg := aq_admin.aq_msg_type (11,'-','-','p1','content priority 1',sysdate);
  v_message_properties.priority := 1; --该消息的优先级别
  v_enqueue_options.visibility :=dbms_aq.immediate;
  dbms_aq.enqueue(
      queue_name => v_queue_name
      ,enqueue_options => v_enqueue_options
      ,message_properties => v_message_properties
      ,payload => v_aq_msg
      ,msgid => v_message_handle);
  dbms_output.put_line('encode success,msgid is ' || v_message_handle);
  commit;
  
  dbms_lock.sleep(1);
  v_aq_msg := aq_admin.aq_msg_type (13,'-','-','p3','content priority 3',sysdate);
  v_message_properties.priority := 3; --该消息的优先级别
  v_enqueue_options.visibility :=dbms_aq.immediate;
  dbms_aq.enqueue(
      queue_name => v_queue_name
      ,enqueue_options => v_enqueue_options
      ,message_properties => v_message_properties
      ,payload => v_aq_msg
      ,msgid => v_message_handle);
  dbms_output.put_line('encode success,msgid is ' || v_message_handle);
  commit;
  
  dbms_lock.sleep(1);
  v_aq_msg := aq_admin.aq_msg_type (12,'-','-','p2','content priority 2',sysdate);
  v_message_properties.priority := 2; --该消息的优先级别
  v_enqueue_options.visibility :=dbms_aq.immediate;
  dbms_aq.enqueue(
      queue_name => v_queue_name
      ,enqueue_options => v_enqueue_options
      ,message_properties => v_message_properties
      ,payload => v_aq_msg
      ,msgid => v_message_handle);
  dbms_output.put_line('encode success,msgid is ' || v_message_handle);
  commit;
end;

7.验证数据 (由下面数据可得知，因会及时回调，所以先入队列的先出对，优先级设置无效）

select t.aq_msg_seq,t.queue_name,t.received_time,t.consumer_name,t.content.msg_content,t.content.msg_create_time  
from aq_msg_received t
---------------------------------------------------------------------------------------------------
AQ_MSG_SEQ  QUEUE_NAME                  RECEIVED_TIME       CONSUMER_NAME   CONTENT.MSG_CONTENT CONTENT.MSG_CREATE_TIME
5           "AQ_ADMIN"."AQ_MSG_QUEUE3"  2020/11/13 22:54:30 SUBSCRIBER_XAG3 content priority 1  2020/11/13 22:54:30
6           "AQ_ADMIN"."AQ_MSG_QUEUE3"  2020/11/13 22:54:31 SUBSCRIBER_XAG3 content priority 3  2020/11/13 22:54:31
7           "AQ_ADMIN"."AQ_MSG_QUEUE3"  2020/11/13 22:54:32 SUBSCRIBER_XAG3 content priority 2  2020/11/13 22:54:32