如何创建基本的高级队列之一：创建发送方代码

/*

基本组成

发送方（DB1）：

Queue type ：决定发送消息的类型

Queue table ：消息发送的载体

Queue ：队列

Subscriber ：订购者，同一个队列可以有n个订购者

Propagation ：传播进程

*/

 

--发送方（DB1 ）代码：

--用账户 SYS 登录

--1.创建 AQ_SENDER 用户并赋权

 

create user AQ_SENDER identified by "oppo";

grant connect,resource,aq_administrator_role,unlimited tablespace to phs;

grant create database link to phs;

grant execute on dbms_aq to phs;

grant execute on dbms_aqadm to phs;

begin

dbms_aqadm.grant_system_privilege('ENQUEUE_ANY', 'phs', FALSE);

dbms_aqadm.grant_system_privilege('DEQUEUE_ANY', 'phs', FALSE);

end;

 

--用账户 AQ_SENDER 登录

--2.创建 db link

create database link MES_LINK

  connect to WMS_APP identified by "oppo"

  using '(DESCRIPTION = (ADDRESS = (PROTOCOL = TCP)(HOST = 10.0.0.115)(PORT = 1521))(CONNECT_DATA = (SID = testassy)))';

 

--3.创建一个队列载体对象，一个没有 body 的 type 在 AQ_SENDER 下

create or replace type TS_BILL_TRANSFER as object

(

  "BILL_NO" VARCHAR2(30),

"LINE_NUMBER" NUMBER,

"STOCK_NO" VARCHAR2(10),

"TO_STOCK_NO" VARCHAR2(10),

"LOCATION" VARCHAR2(60),

"PART_NO" VARCHAR2(60),

"PART_DESC" VARCHAR2(240),

"PART_QTY" NUMBER,

"OPERATOR" VARCHAR2(30),

"DATETIME_CREATED" DATE

);

 

--4.创建队列表

begin

-- Call the procedure

sys.dbms_aqadm.create_queue_table(queue_table => 'QT_BILL_TRANSFER',

          queue_payload_type => 'TS_BILL_TRANSFER',--这就是我们定义的 type

          sort_list => 'priority,enq_time',--按优先级和入列时间排序

          multiple_consumers => TRUE, --多消费者

          comment => 'QUEUE FOR SEND BILL-TRANSFER DATA TO ERP',

          auto_commit => FALSE --手动控制事务--create queue

          );

end;

 

--5.创建队列

begin

sys.dbms_aqadm.create_queue(

queue_name => 'Q_BILL_TRANSFER',

queue_table => 'QT_BILL_TRANSFER',--刚刚建立的queue表

queue_type => sys.dbms_aqadm.normal_queue,

max_retries => 3,--dequeue失败后重试次数

retry_delay => 1,--重试前等待

retention_time => 0 --dequeue后保持时间，不保持

);

end;

 

--6.启动队列

begin

dbms_aqadm.start_queue('Q_BILL_TRANSFER',true,true);

end;

 

--7.创建消息订阅者

DECLARE

subscriber sys.aq$_agent;

BEGIN

subscriber := sys.aq$_agent('PHS', 'PHS.Q_BILL_TRANSFER', NULL);

DBMS_AQADM.ADD_SUBSCRIBER(queue_name => 'Q_BILL_TRANSFER', subscriber => subscriber);

--DBMS_AQADM.remove_subscriber(queue_name => 'Q_BILL_TRANSFER', subscriber => subscriber);

END;

 

--8.创建 propagation

BEGIN

DBMS_AQADM.SCHEDULE_PROPAGATION(queue_name => 'Q_BILL_TRANSFER', destination => 'MES_LINK');

--同一数据库，不同用户

--DBMS_AQADM.SCHEDULE_PROPAGATION(queue_name => 'Q_BILL_TRANSFER', destination => '');

END;

 

-----------------------------------------------------------------------------------------------

-----------------------------------------------------------------------------------------------

--9.创建接收方（DB2 ）相关代码，请打开文件 CODE_RECEIVER.SQL 在接收方（DB2 ）数据库执行

-----------------------------------------------------------------------------------------------

-----------------------------------------------------------------------------------------------

 

--10.入列测试（该段代码每执行一次，只导入一条 Queue 数据）

declare

v_Message TS_BILL_TRANSFER;

v_MsgId RAW(16);

v_options DBMS_AQ.ENQUEUE_OPTIONS_T;

v_properties DBMS_AQ.MESSAGE_PROPERTIES_T;

v_Recipients DBMS_AQ.AQ$_RECIPIENT_LIST_T;

begin

  v_Message:=TS_BILL_TRANSFER(

  BILL_NO => 'B001',

LINE_NUMBER => 003,

STOCK_NO => '327',

TO_STOCK_NO => '303',

LOCATION  => 'B378',

PART_NO => '1100825',

PART_DESC => '扬声器 0.5W 8Ω 15×11×3.5 弹片 4#',

PART_QTY => 35,

OPERATOR => 'admin',

DATETIME_CREATED=> SYSDATE);

  v_properties.priority := 1; --该消息的优先级别

  v_options.visibility :=DBMS_AQ.IMMEDIATE;

  dbms_aq.enqueue(queue_name => 'Q_BILL_TRANSFER',enqueue_options => v_options,message_properties =>

  v_properties, payload => v_Message, msgid => v_MsgId);

  dbms_output.put_line('encode success,msgid is '||v_MsgId);

end;

 

--查询QTABLE

SELECT * FROM QT_BILL_TRANSFER;

--查询操作的数据

SELECT * FROM INV_SP_BILL_TRANSFER;

 

SELECT * FROM Q_TEST;

 

begin

  -- Call the procedure

  q_sp_test;

end;

 

 

--QUEUE 相关查询

select * from user_queue_tables;

select * from user_queues;

select * from user_queue_subscribers;

select * from user_queue_schedules;

SELECT * FROM USER_QUEUE_PUBLISHERS;