Advanced Queueing

From Oracle FAQ
Jump to: navigation, search

Oracle Advanced Queueing (AQ) is the Oracle database's queue management feature. AQ provides a message queuing infrastructure as integral part of the Oracle server engine. It provides an API for enqueing messages to database queues. These messages can later be dequeued for asynchronous processing. Oracle AQ also provides functionality to preserve, track, document, correlate, and query messages in queues.

History[edit]

Oracle AQ was first introduced in Oracle 8.

As of Oracle release 9.2, AQ is bundled with Standard Edition and Enterprise Edition at no extra cost.

As of Oracle release 10.1, AQ is integrated into Oracle Streams, and is called "Oracle Streams AQ".

Interfaces[edit]

Oracle AQ is available from within PL/SQL programs through the DBMS_AQADM and DBMS_AQ built-in packages.

Some of the Oracle features that utilize AQ:

With Data Guard, the primary database's queue monitor (qmn0) process interacts with AQ.

Oracle AQ is used as the internal Java Message Service provider in the Oracle Enterprise Service Bus.

Example[edit]

Prepare database user:

CONN / AS SYSDBA
CREATE USER testq IDENTIFIED BY x;
GRANT connect, resource, dba TO testq;
GRANT aq_administrator_role, aq_user_role  TO testq;
GRANT create type TO testq;

Create a queue:

CONN testq/x@ora10gtest
SET SERVEROUTPUT ON

CREATE OR REPLACE TYPE event_msg_type AS OBJECT (
  name            VARCHAR2(10),
  current_status  NUMBER(5),
  next_status     NUMBER(5)
);
/

EXECUTE DBMS_AQADM.create_queue_table( -
   queue_table        =>  'testq.event_queue_tab', -
   queue_payload_type =>  'testq.event_msg_type');

EXECUTE DBMS_AQADM.create_queue( -
   queue_name         =>  'testq.event_queue', -
   queue_table        =>  'testq.event_queue_tab');

EXECUTE DBMS_AQADM.start_queue( -
   queue_name         => 'testq.event_queue', -
   enqueue            => TRUE);

Test an enqueue...

DECLARE
  l_enqueue_options    DBMS_AQ.enqueue_options_t;
  l_message_properties DBMS_AQ.message_properties_t;
  l_message_handle     RAW(16);
  l_event_msg          TESTQ.event_msg_type;
BEGIN
  l_event_msg := TESTQ.event_msg_type('REPORTER', 1, 2);
  DBMS_AQ.enqueue(queue_name         => 'testq.event_queue',
                  enqueue_options    => l_enqueue_options,
                  message_properties => l_message_properties,
                  payload            => l_event_msg,
                  msgid              => l_message_handle);
  COMMIT;
END;
/

Test a dequeue:

DECLARE
  l_dequeue_options    DBMS_AQ.dequeue_options_t;
  l_message_properties DBMS_AQ.message_properties_t;
  l_message_handle     RAW(16);
  l_event_msg          TESTQ.event_msg_type;
BEGIN
  DBMS_AQ.dequeue(queue_name         => 'testq.event_queue',
                  dequeue_options    => l_dequeue_options,
                  message_properties => l_message_properties,
                  payload            => l_event_msg,
                  msgid              => l_message_handle);
  DBMS_OUTPUT.put_line('Event Name    : ' || l_event_msg.name);
  DBMS_OUTPUT.put_line('Current Status: ' || l_event_msg.current_status);
  DBMS_OUTPUT.put_line('Next Status   : ' || l_event_msg.next_status);
  COMMIT;
END;
/

External links[edit]