--author xujun 整理
--oracle 10G及以上
--1、以管理员登录
conn sys/password@orcl as sysdba;

--设置JOB进程数 如10 或15 以保证数据库可以启动JOB进程
alter system set job_queue_processes = 10; 

--2、建立用户
--drop user aq_admin CASCADE;
create user aq_admin identified by aq_admin
default tablespace users
temporary tablespace temp
quota unlimited on users;
--3、给用户付权
grant connect,
      create type,
      create procedure,
      create table,
      aq_administrator_role
   to aq_admin;

grant execute on dbms_aq to aq_admin;
grant execute on dbms_aqadm to aq_admin;

/*--权限
begin
dbms_aqadm.grant_system_privilege('ENQUEUE_ANY','AQ',FALSE);
dbms_aqadm.grant_system_privilege('DEQUEUE_ANY','AQ',FALSE);
end;
/*/

--4、以aq_admin用户登录
connect aq_admin/aq_admin@orcl;
--5、建立队列TYPE
create type TYP_DB_TASK as object
(
  SUBJECT varchar2(30),
  TEXT    varchar2(80),
  CREATE_DATE DATE
);
/
--6、建立队列表、队列、启动队列
begin
 DBMS_AQADM.CREATE_QUEUE_TABLE(QUEUE_TABLE => 'QT_DB_TASK',
                QUEUE_PAYLOAD_TYPE => 'TYP_DB_TASK',
                MULTIPLE_CONSUMERS => FALSE);

 DBMS_AQADM.CREATE_QUEUE(QUEUE_NAME => 'Q_DB_TASK',
             QUEUE_TABLE => 'QT_DB_TASK');

 DBMS_AQADM.START_QUEUE(QUEUE_NAME => 'Q_DB_TASK');
end;
/

--7、建立入队过程
create or replace procedure PRC_TASK_ENQUEUE(AC_MSG in varchar2) as
 VT_ENQUEUE_OPTIONS    DBMS_AQ.ENQUEUE_OPTIONS_T;
 VT_MESSAGE_PROPERTIES DBMS_AQ.MESSAGE_PROPERTIES_T;
 VR_MESSAGE_HANDLE     raw(16);
 VT_MESSAGE            TYP_DB_TASK;
begin
 VT_MESSAGE := TYP_DB_TASK('NORMAL TASK', AC_MSG, sysdate);

 DBMS_AQ.ENQUEUE(QUEUE_NAME => 'Q_DB_TASK',
         ENQUEUE_OPTIONS => VT_ENQUEUE_OPTIONS,
         MESSAGE_PROPERTIES => VT_MESSAGE_PROPERTIES,
         PAYLOAD => VT_MESSAGE, MSGID => VR_MESSAGE_HANDLE);
 commit;
exception
 when others then
  DBMS_OUTPUT.PUT_LINE('Can''t enqueue.');
end;
/

--8、建立任务消息日志表
--drop table S_DB_TASK_LOG;
create table S_DB_TASK_LOG(
  TITLE VARCHAR2(100),
  MSG TYP_DB_TASK
);

--9、建立响应队列通知事件的过程，执行出队操作，执行任务过程，记录任务日志
create or replace procedure PRC_NOTIFYCB
(
 context  raw,
 REGINFO  SYS.AQ$_REG_INFO,
 DESCR    SYS.AQ$_DESCRIPTOR,
 PAYLOAD  raw,
 PAYLOADL number
) as
 VT_DEQUEUE_OPTIONS    DBMS_AQ.DEQUEUE_OPTIONS_T;
 VT_MESSAGE_PROPERTIES DBMS_AQ.MESSAGE_PROPERTIES_T;
 VR_MESSAGE_HANDLE     raw(16);
 VT_MESSAGE            TYP_DB_TASK;
begin
 VT_DEQUEUE_OPTIONS.MSGID := DESCR.MSG_ID;
 VT_DEQUEUE_OPTIONS.CONSUMER_NAME := DESCR.CONSUMER_NAME;

 DBMS_AQ.DEQUEUE(QUEUE_NAME => DESCR.QUEUE_NAME,
         DEQUEUE_OPTIONS => VT_DEQUEUE_OPTIONS,
         MESSAGE_PROPERTIES => VT_MESSAGE_PROPERTIES,
         PAYLOAD => VT_MESSAGE, MSGID => VR_MESSAGE_HANDLE);

 insert into S_DB_TASK_LOG
 values
  ('Dequeued and processed',
   VT_MESSAGE);
 commit;

 --执行具体任务过程
 --略
exception
 when others then
  DBMS_OUTPUT.PUT_LINE('Can''t dequeue.');
end;
/

--10、增加订阅者,用于多订阅者情况，前面MULTIPLE_CONSUMERS要设为true
/*
begin
 DBMS_AQADM.ADD_SUBSCRIBER(QUEUE_NAME => 'Q_DB_TASK',
              SUBSCRIBER => SYS.AQ$_AGENT('DB_RECIPIENT', null, null));
end;
/
*/

--11、注册响应队列通知事件的过程
begin
 DBMS_AQ.REGISTER(SYS.AQ$_REG_INFO_LIST(SYS.AQ$_REG_INFO('Q_DB_TASK',
                             DBMS_AQ.NAMESPACE_AQ,
                             'plsql://prc_notifycb',
                             HEXTORAW('FF'))), 1);
end;
/
--取消注册
/*
begin
 DBMS_AQ.UNREGISTER(SYS.AQ$_REG_INFO_LIST(SYS.AQ$_REG_INFO('Q_DB_TASK',
                              DBMS_AQ.NAMESPACE_AQ,
                              'plsql://prc_notifycb',
                              HEXTORAW('FF'))), 1);
end;
/*/
--12、测试，快速执行以下
exec prc_task_enqueue ('异步并发过程1');
exec prc_task_enqueue ('异步并发过程2');
exec prc_task_enqueue ('异步并发过程3');
exec prc_task_enqueue ('异步并发过程4');

--13、After a short while
select * from S_DB_TASK_LOG;
select * from QT_DB_TASK;