Oracle AQ队列 管理应用程序流程的关键工具(oracle aq 队列)
  iDU31ygkXmx7 2023年11月09日 2 0

Oracle AQ队列: 管理应用程序流程的关键工具

Oracle Advanced Queueing(AQ)是Oracle数据库提供的一种基于消息传递机制的队列服务,它支持异步消息处理,可以帮助应用程序处理大量的事务性数据和并发请求,并提供数据传递的可靠性和可恢复性。本文将介绍Oracle AQ队列的基本概念和使用方法,并通过示例代码演示如何在应用程序中使用Oracle AQ队列。

一、Oracle AQ队列的基本概念

Oracle AQ队列是一个在Oracle数据库中创建的对象,它由3部分组成:队列表、队列、和消息。队列表是支持AQ队列的Oracle表,它包含队列的定义和属性以及队列中消息的存储。队列是在队列表上创建的一个或多个逻辑队列,每个队列有唯一的名称和属性。消息是放置在队列中的传递对象,它由消息体和属性组成。队列中的消息可以由生产者发送并由消费者接收和处理。

Oracle AQ队列提供了以下基本功能:

(1) 消息的持久性:可以将消息保存到数据库中以实现数据的可靠性和持久性。

(2) 异步消息传递:可以在不等待响应的情况下发送和接收消息。

(3) 事务处理:可以将消息的发送和接收与数据库事务进行绑定,以确保数据的一致性和可靠性。

(4) 消息优先级:可以为消息设置不同的优先级,以确保高优先级消息被及时处理。

二、Oracle AQ队列的使用方法

创建队列表

在Oracle数据库中创建队列表的方法如下:

CREATE TABLE queue_table_name

(queue_column_name queue_type, …)

LOB(queue_column_name) STORE AS

(queue_type LOB_type)

TABLESPACE tablespace_name;

其中,queue_table_name是队列表的名称,queue_column_name是队列表中的列名,queue_type是Oracle AQ预定义的数据类型(如sys.aq$_jms_text_message、sys.aq$_jms_map_message等),LOB_type是大对象的类型(如CLOB、BLOB等),tablespace_name是队列表所在的表空间名称。

创建队列

在Oracle数据库中创建队列的方法如下:

BEGIN

DBMS_AQADM.CREATE_QUEUE(queue_name => ‘queue_name’,

queue_table => ‘queue_table_name’,

queue_type => sys.aq$_jms_text_message);

END;

其中,queue_name是队列的名称,queue_table_name是队列表的名称,sys.aq$_jms_text_message是队列中消息的类型。

发送消息

在Oracle数据库中发送消息的方法如下:

DECLARE

msg sys.aq$_jms_text_message;

propid NUMBER;

qn sys.aq$_agent;

DEQUEUE_OPTIONS sys.aq$_dequeue_options;

BEGIN

broker_options.msgproperties :=

sys.aq$_jms_header_property(‘PRIORITY’, ‘1’);

msg := sys.aq$_jms_text_message.init(‘Hello AQ!’);

propid := broker_options.msgproperties.add_property(‘id’, ‘1’);

qn := sys.aq$_agent.init(‘queue_owner.queue_name’,

NULL,

‘queue_owner’);

DBMS_AQ.ENQUEUE(queue_name => ‘queue_owner.queue_name’,

enqueue_options => NULL,

message_properties => broker_options.msgproperties,

payload => msg,

msgid => propid,

msg_correlation => NULL,

transform => NULL,

queue_agent => qn);

END;

其中,queue_owner是队列的所有者,queue_name是队列的名称,broker_options.msgproperties是消息的属性,msg是消息的内容。

接收消息

在Oracle数据库中接收消息的方法如下:

DECLARE

msg sys.aq$_jms_text_message;

propid NUMBER;

qn sys.aq$_agent;

DEQUEUE_OPTIONS sys.aq$_dequeue_options;

BEGIN

qn := sys.aq$_agent.init(NULL,

‘queue_owner.queue_name’,

‘queue_owner’);

DEQUEUE_OPTIONS.CONSUMER_NAME := ‘consumer_name’;

DEQUEUE_OPTIONS.NAV_MODE := DBMS_AQ.FIRST_MESSAGE;

DBMS_AQ.DEQUEUE(queue_name => ‘queue_owner.queue_name’,

dequeue_options => DEQUEUE_OPTIONS,

message_properties => NULL,

payload => msg,

msgid => propid,

msg_correlation => NULL,

transform => NULL,

queue_agent => qn);

END;

其中,queue_owner是队列的所有者,queue_name是队列的名称,consumer_name是消费者的名称,msg是接收到的消息的内容。

三、示例代码

下面是一个简单的示例,在这个示例中,我们将创建一个队列和2个消费者,每个消费者都会向队列中发送5条消息,每条消息包含一个递增的整数值。第三个消费者会从队列中接收并打印所有的消息。

1. 创建队列

BEGIN

DBMS_AQADM.CREATE_QUEUE(queue_name => ‘test_queue’,

queue_table => ‘test_queue_table’,

queue_type => sys.aq$_jms_text_message);

DBMS_AQADM.START_QUEUE(queue_name => ‘test_queue’);

END;

2. 创建消费者

CREATE OR REPLACE PROCEDURE consumer1_proc AS

queue_handle SYS.AQ$_QUEUE;

message_handle SYS.AQ$_AGENT;

enq_options DBMS_AQ.ENQUEUE_OPTIONS_T;

propid NUMBER;

msg sys.aq$_jms_text_message;

BEGIN

message_handle := SYS.AQ$_AGENT(‘test_queue_owner’, ‘test_queue_name’, NULL);

queue_handle := SYS.AQ$_QUEUE(message_handle, ‘test_queue_owner’, ‘test_queue_name’);

FOR i IN 1..5 LOOP

msg := sys.aq$_jms_text_message.init(‘Message ‘ || TO_CHAR(i) || ‘ from consumer 1!’);

propid := enq_options.msgproperties.add_property(‘id’, i);

DBMS_AQ.ENQUEUE(

queue_name => queue_handle,

enqueue_options => enq_options,

message_properties => enq_options.msgproperties,

payload => msg,

msgid => propid);

END LOOP;

END;

CREATE OR REPLACE PROCEDURE consumer2_proc AS

queue_handle SYS.AQ$_QUEUE;

message_handle SYS.AQ$_AGENT;

enq_options DBMS_AQ.ENQUEUE_OPTIONS_T;

propid NUMBER;

msg sys.aq$_jms_text_message;

BEGIN

message_handle := SYS.AQ$_AGENT(‘test_queue_owner’, ‘test_queue_name’, NULL);

queue_handle := SYS.AQ$_QUEUE(message_handle, ‘test_queue_owner’, ‘test_queue_name’);

FOR i IN 1..5 LOOP

msg := sys.aq$_jms_text_message.init(‘Message ‘ || TO_CHAR(i) || ‘ from consumer 2!’);

propid := enq_options.msgproperties.add_property(‘id’, i);

DBMS_AQ.ENQUEUE(

queue_name => queue_handle,

enqueue_options => enq_options,

message_properties => enq_options.msgproperties,

payload => msg,

msgid => propid);

END LOOP;

END;

3. 接收消息

CREATE OR REPLACE PROCEDURE consumer3_proc AS

queue_handle SYS.AQ$_QUEUE;

message_handle SYS.AQ$_AGENT;

deq_options DBMS_AQ.DEQUEUE_OPTIONS_T;

buf sys.DBMS_SQL.VARCHAR2A;

n INTEGER;

msg sys.aq$_jms_text_message;

BEGIN

message_handle := SYS.AQ$_AGENT(NULL, ‘test_queue_owner’, ‘test_queue_name’);

queue_handle := SYS.AQ$_QUEUE(message_handle, ‘test_queue_owner’, ‘test_queue_name’);

deq_options.consumer_name := ‘consumer3’;

deq_options.navigation_mode := DBMS_AQ.FIRST_MESSAGE;

deq_options.visibility := DBMS_AQ.IMMEDIATE;

n := 0;

LOOP

DBMS_AQ.DEQUEUE(

queue_name => queue_handle,

dequeue_options => deq_options,

message_properties => NULL,

payload => msg,

msgid => buf(n),

msg_correlation => NULL);

n := n+1;

DBMS_OUTPUT.PUT_LINE(msg.text || ‘ received by consumer 3’);

DEQ_OPTIONS.MSGID := buf(n);

EXCEPTION

WHEN NO_DATA_FOUND THEN

IF n = 0 THEN

DBMS_OUTPUT.PUT_LINE(‘Queue is empty…’);

ELSE

DBMS_OUTPUT.PUT_LINE(‘All messages received!’);

END IF;

EXIT;

END;

END LOOP;

END;

执行consumer1_proc、consumer2_proc、consumer3_proc后,可以看到如下输出:

Message 1 from consumer 1! received by consumer 3

Message 1 from consumer 2! received by consumer 3

Message 2 from consumer 1! received by consumer 3

Message 2 from consumer 2! received by consumer 3

Message 3 from consumer 1! received by consumer 3

Message 3 from consumer 2! received by consumer 3

Message 4 from consumer 1! received by consumer 3

Message 4 from consumer 2! received by consumer 3

Message 5 from consumer 1! received by consumer 3

Message 5 from consumer

【版权声明】本文内容来自摩杜云社区用户原创、第三方投稿、转载,内容版权归原作者所有。本网站的目的在于传递更多信息,不拥有版权,亦不承担相应法律责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@moduyun.com

  1. 分享:
最后一次编辑于 2023年11月09日 0

暂无评论

推荐阅读
iDU31ygkXmx7