Advanced Queuing and Transactional Event Queues

Oracle Database Advanced Queuing (AQ) and Oracle Transactional Event Queues (TxEventQ) provide database-integrated message queuing functionality. AQ leverages the functions of Oracle Database so that messages can be stored persistently, propagated between queues on different computers and databases, and transmitted using Oracle Net Services and HTTP(S). TxEventQ provides robust real-time messaging, streaming events, and publish/subscribe messaging with multiple publishers and multiple consumers. It is a high performance partitioned implementation with multiple event streams per queue, while AQ is a disk-based implementation for simpler workflow use cases.

ODP.NET uses the same APIs when executing operations against either AQ or TxEventQ. From a .NET application level, both queue types have the same functionality. Thus, ODP.NET developers can use the same code for working with either AQ or TxEventQ. Where the two queue types differ is in their implementations and administration.

Due to the commonality at the .NET app level, this book will use the term “AQ” to refer synonomously to both AQ and TxEventQ. When differences exist, the book will refer to each queue type separately.

ODP.NET queuing APIs execute operational functions, such as enqueuing and dequeuing messages. Administrative functions, such as creating and managing queues, are not available in ODP.NET. Use Oracle Developer Tools for Visual Studio or DBMS_AQADM PL/SQL package to perform these administrative functions.

As Oracle Database AQ is implemented in database tables, all operational benefits of high availability, scalability, and reliability are also applicable to queue data. Oracle Database AQ supports standard database features such as recovery, restart, and security.

The following items discuss Oracle Database AQ concepts:

See Also:

Using ODP.NET for Advanced Queuing

.NET applications can use ODP.NET to access all the operational features of AQ such as Enqueuing, Dequeuing, Listen, and Notification. Table 3-38 maps the AQ features to their corresponding ODP.NET implementation.

Table 38 Mapping AQ Features with their ODP.NET Implementation

Functionality ODP.NET Implementation
Create a Message Create an OracleAQMessage object
Enqueue a single message Specify the message as OracleAQMessage, queue as OracleAQQueue and enqueue options on OracleAQQueue, call OracleAQQueue.Enqueue
Enqueue multiple messages Specify the messages as an OracleAQMessage array in OracleAQQueue.EnqueueArray
Dequeue a single message Specify dequeue options on OracleAQQueue and call OracleAQQueue.Dequeue
Dequeue multiple messages Call OracleAQQueue.DequeueArray
Listen for messages on Queue(s) Call OracleAQQueue.Listen.To listen on multiple queues use static Listen method of OracleAQQueue
Message Notifications Use OracleAQQueue.MessageAvailable Event along with the NotificationConsumers property

Enqueuing and Dequeuing Example

The following example demonstrates enqueuing and dequeuing messages using a single consumer queue. The first part of the example performs the requisite database setup for the database user, SCOTT. The second part of the example demonstrates enqueuing and dequeuing messages.

-- Part I: Database setup required for this demo

------------------------------------------------------------------
-- SQL to grant appropriate privilege to database user, SCOTT
------------------------------------------------------------------
SQL> ALTER USER SCOTT ACCOUNT UNLOCK IDENTIFIED BY Pwd4Sct;
User altered.
SQL> GRANT ALL ON DBMS_AQADM TO scott;

------------------------------------------------------------------
-- PL/SQL to create queue-table and queue and start queue for SCOTT
------------------------------------------------------------------
BEGIN
  DBMS_AQADM.CREATE_QUEUE_TABLE(
    queue_table=>'scott.test_q_tab',
    queue_payload_type=>'RAW',
    multiple_consumers=>FALSE);

  DBMS_AQADM.CREATE_QUEUE(
    queue_name=>'scott.test_q',
    queue_table=>'scott.test_q_tab');

  DBMS_AQADM.START_QUEUE(queue_name=>'scott.test_q');
END;
/

------------------------------------------------------------------
-- PL/SQL to stop queue and drop queue & queue-table from SCOTT
------------------------------------------------------------------
BEGIN
  DBMS_AQADM.STOP_QUEUE('scott.test_q');

  DBMS_AQADM.DROP_QUEUE(
    queue_name => 'scott.test_q',
    auto_commit => TRUE);

  DBMS_AQADM.DROP_QUEUE_TABLE(
    queue_table => 'scott.test_q_tab',
    force => FALSE,
    auto_commit => TRUE);
END;
/
-- End of Part I, database setup.

//Part II: Enqueuing and dequeuing messages
//C#
using System;
using System.Text;
using Oracle.DataAccess.Client;
using Oracle.DataAccess.Types;

namespace ODPSample
{
  /// <summary>
  /// Demonstrates Enqueuing and Dequeuing raw message
  /// using a single consumer queue
  /// </summary>
  class EnqueueDequeue
  {
    static void Main(string[] args)
    {
      // Create connection
      string constr = "user id=scott;password=Pwd4Sct;data source=oracle";
      OracleConnection con = new OracleConnection(constr);

      // Create queue
      OracleAQQueue queue = new OracleAQQueue("scott.test_q", con);

      try
      {
        // Open connection
        con.Open();

        // Begin txn for enqueue
        OracleTransaction txn = con.BeginTransaction();

        // Set message type for the queue
        queue.MessageType = OracleAQMessageType.Raw;

        // Prepare message and RAW payload
        OracleAQMessage enqMsg = new OracleAQMessage();
        byte[] bytePayload = { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
        enqMsg.Payload = bytePayload;

        // Prepare to Enqueue
        queue.EnqueueOptions.Visibility = OracleAQVisibilityMode.OnCommit;

        // Enqueue message
        queue.Enqueue(enqMsg);

        Console.WriteLine("Enqueued Message Payload      : "
          + ByteArrayToString(enqMsg.Payload as byte[]));
        Console.WriteLine("MessageId of Enqueued Message : "
          + ByteArrayToString(enqMsg.MessageId));

        // Enqueue txn commit
        txn.Commit();

        // Begin txn for Dequeue
        txn = con.BeginTransaction();

        // Prepare to Dequeue
        queue.DequeueOptions.Visibility = OracleAQVisibilityMode.OnCommit;
        queue.DequeueOptions.Wait = 10;

        // Dequeue message
        OracleAQMessage deqMsg = queue.Dequeue();

        Console.WriteLine("Dequeued Message Payload      : "
          + ByteArrayToString(deqMsg.Payload as byte[]));
        Console.WriteLine("MessageId of Dequeued Message : "
          + ByteArrayToString(deqMsg.MessageId));

        // Dequeue txn commit
        txn.Commit();
      }
      catch (Exception e)
      {
        Console.WriteLine("Error: {0}", e.Message);
      }
      finally
      {
        // Close/Dispose objects
        queue.Dispose();
        con.Close();
        con.Dispose();
      }
    }

    // Function to convert byte[] to string
    static private string ByteArrayToString(byte[] byteArray)
    {
      StringBuilder sb = new StringBuilder();
      for (int n = 0; n < byteArray.Length; n++)
      {
        sb.Append((int.Parse(byteArray[n].ToString())).ToString("X"));
      }
      return sb.ToString();
    }
  }
}