Advanced Queuing Technology

In This Topic

Advanced Queuing (AQ) is a flexible message exchange mechanism built in Oracle server. With AQ you can send messages from a workstation or server to one or more workstations. dotConnect for Oracle incorporates set of classes that leverage working with this technology.

This feature is available in Professional and Developer Editions only.

This topic provides information and examples that can help you start with AQ. It consists of the following sections:

AQ Basics
AQ components in dotConnect for Oracle
A Simple Example of Point-to-Point Messaging
An Extended Example of Point-to-Point Messaging
A Simple Example of Point-to-Multipoint Messaging

AQ Basics

Advanced Queuing provides database-integrated message queuing functionality. Advanced Queue messages can be stored persistently, propagated between queues on different machines and databases, and transmitted using Oracle Net Services, HTTP(S), and SMTP. Since Oracle Advanced Queuing is implemented in database tables, all the benefits of high availability, scalability, and reliability are applicable to queue data. This technology is presented by DBMS_AQ and DBMS_AQADM packages. DBMS_AQ manages queue messages, allowing enqueuing and dequeuing. DBMS_AQADM controls a queue lifecycle and its properties.

Standard database features such as recovery, restart, and security are supported in Advanced Queuing, and queue tables can be imported and exported. Advanced Queuing provides the message management functionality and asynchronous communication needed for application integration.

Advanced Queuing is a complicated area, so it is recommended that you read the official documentation for better understanding of AQ aspects and capabilities.

There are two models of AQ functionality: Point-to-Point Model and Publish-Subscribe Model. In the first one, there is an application which puts messages to the queue (this is called enqueuing) and a consumer application which uses enqueued messages, removing them from the queue (this is called dequeuing). There can be more than one consumer applications but any single message can be dequeued just once. Also, consumers may browse the queue without dequeuing messages.

In the more sophisticated Publish-Subscribe model we have some applications (publishers) that enqueue messages and some others (subscibers) that dequeue these messages. Publishers propagate messages to queues which are called topics here. These messages can be either addressed for specific applications or they will be received by all destinations. Applications receiving messages are also called agents. We talk about broadcast if the message can be consumed by all applications and about multicast if a subscription is required for consummation.

AQ components in dotConnect for Oracle

There are a number of classes used to provide the AQ functionality in dotConnect for Oracle. The main of them are:

The OracleQueueMessage objects represent queue messages. They are used as arguments when enqueuing messages and are returned by dequeue methods. OracleQueueMessages contain the system-generated message identificator and some message payload, which is of either char or user-defined object type.

The OracleQueue class provides the main queue functionality, such as enqueuing and dequeuing messages. The Enqueue method of OracleQueue take as arguments either OracleQueueMessage objects or any data sufficient for creating the message. In both cases it is possible to enqueue the array of messages via the EnqueueArray method. Dequeuing returns either single OracleQueueMessage (the Dequeue method) or their array (the DequeueArray method). The OracleQueue object can be limited to either enqueuing or dequeuing only. This option may be useful e.g. when one application only generates messages and the other one only consumes them; in this case the first application can use the OracleQueue instance which enqueues messages, while the second application will have the OracleQueue instance limited for dequeuing only.

To administrate the queues itself the OracleQueueAdmin class is used. It allows to create/drop and start/stop message queues as well as to set up their numerous properties. In Publish-Subscribe model, OracleQueueAdmin also manages the queue subscribers.

The queues can consist of either persistent or buffered messages. Persistent messages are stored in the database, buffered messages are not; hence, buffered messages are lost when the application shuts down. Persistent queues need special tables in the database; the functionality needed to create and manage such tables is provided by the OracleQueueTable class.

In Publish-Subscribe model, the consumers and publishers are represented by the OracleQueueAgent class. The specific consumer needs to be subscribed to the queue to get its messages, this can be made using OracleQueueAdmin. The publishers are referred in the properties of every message sent.

There are also special classes representing the property sets of messages, queues etc. These property classes are:

OracleQueueOptions representing the queue properties which concern storing and managing messages.
OracleQueueEnqueueOptions and OracleQueueDequeueOptions corresponding to the settings of enqueue/dequeue operations. Unless there are some options specified for the particular message, it is enqueued or dequeued according to OracleQueueEnqueueOptions or OracleQueueDequeueOptions.
OracleQueueMessageProperties class is used to set the properties of individual messages.
OracleQueueTableOptions represents the queue table settings such as compatibility settings, message payload types and message grouping properties.

A Simple Example of Point-to-Point Messaging

The following example demonstrates how to use Advanced Queueing in its simplest form. It represents a point-to-point messaging scheme where messages are deleted from server once they are consumed by the recipient.

First of all, we need to create a connection to the Oracle database. We are going to use object payload for messages:

Sometimes you need messages of some user-defined type (UDT) to contain at least a title or identification number and some content, e.g. message text. In this case, we need to specify which type of messages will be used in our AQ sample. All queue messages will be structured as this UDT. The type itself must be defined in the database.

Message queues are stored in special tables. The scripts, needed to create and manipulate such tables, are incapsulated in a single OracleQueueTable class. We will create an instance of this class, set the type of messages used in the queue and create the queue table at the database:

The queue lifecycle is managed by OracleQueueAdmin class. The objects of this class create, start, stop and drop queues. Here we will create a new queue administration object and bind it to the "QUEUE_TABLE_MESSAGE" table. Also, we need to specify the name of the currently managed queue ("MESSAGE_QUEUE"):

Now we are able to create and start the queue itself:

After initializing the queue we can enqueue and dequeue messages. This operation would be more helpful in the case of several different applications, but we'll confine ourselves with the only one to make the sample more clear.

We will create an object of the OracleQueue class on the "publisher"'s side (which is the only one in our case). The OracleQueue class allows to operate with the messages concerned with the current queue. In particular, it allows the "publisher" putting messages to the queue:

The following code creates a message object mes and enqueues the corresponding message:

Proceed to the "consumer"'s side. Create object oracleDequeueQueue of OracleQueue class.

Get the message from the queue. By default the Dequeue() method does not return until it successfully receives a message; in PtP model all messages are deleted from the queue after being dequeued. The behavior of the Dequeue method will be explained in details later.

Stop the queue:

Delete the queue from the associated table. The queue will throw an exception on attempt to drop it, if it isn't stopped.

Drop the queue table. The queue table needs all its queues to be stopped and deleted to be available for dropping. To automatically stop and delete the queues, use the DropQueueTable(bool force) overload.

There are several ways how the OracleQueue.Dequeue() method behaves. It depends on the value of OracleQueueDequeueOptions.WaitTimeout property. When this value is -1 (by default), the Dequeue() method waits for a message to come, and does not return until the message arrives. When the value is 0, the Dequeue() method checks for a message and returns immediately. If there is no message, an exception is thrown. When the value is a positive integer, it designates a number of seconds to wait for a message. If no message is received during this timeout, an exception is thrown.

In the end of the sample we stop and destroy the queue together with its table. In the real world this may be not necessary, as the queue will continue to work after the application that created it is shut down.

An Extended Example of Point-to-Point Messaging

The next example slightly alters the code from the previous section. First, the payload type is changed to RAW to avoid creation of user-defined types for messaging. Second, a priority is assigned to one of the messages sent through the AQ mechanism. Third, messages are received in a more real-world way, with loops and finite timeouts.

OracleConnection oracleConnection = new OracleConnection(
   "User Id=system;Password=manager;Server=ora;");
oracleConnection.Open();

OracleQueueTable oracleQueueTable = new OracleQueueTable(
   "QUEUE_TABLE_MESSAGE", oracleConnection);

// Set sort order by priority for the queue.  
// The messages with higher priority will reach the recipient first.
oracleQueueTable.Options.SortOrder = OracleQueueSortOrder.PriorityEnqueueTime;

// Specify type of the messages in the queue. 
//This time each message will be represented by just a string object. 
//We do so to simplify the unimportant details and concentrate on the advanced features of AQ
	
oracleQueueTable.Options.PayloadTypeName = "RAW";

// The following operations are same as in the previous example.
oracleQueueTable.CreateQueueTable();
OracleQueueAdmin oracleQueueAdmin = new OracleQueueAdmin("MESSAGE_QUEUE", 
"QUEUE_TABLE_MESSAGE", oracleConnection);
oracleQueueAdmin.CreateQueue();
oracleQueueAdmin.StartQueue();
OracleQueue oracleEnqueueQueue = new OracleQueue("MESSAGE_QUEUE", oracleConnection);

// Create and send the first message.
OracleQueueMessage message1 = new OracleQueueMessage();
message1.StringPayload = "First message.";
message1.MessageProperties.Priority = 7;
oracleEnqueueQueue.Enqueue(message1);

// Create and send the second message. This message is assigned a higher priority value.
// The message will be consumed first, regardless of the fact that it was sent later.
OracleQueueMessage message2 = new OracleQueueMessage();
message2.StringPayload = "Second message with high priority.";
message2.MessageProperties.Priority = 1;
oracleEnqueueQueue.Enqueue(message2);

// Create an object that receives the two messages.
OracleQueue oracleDequeueQueue = new OracleQueue("MESSAGE_QUEUE", oracleConnection);
oracleDequeueQueue.DequeueOptions.WaitTimeout = 1;

// Retrieve the messages in a loop. Once there are two messages received, quit.
// (ex.Code == 25228) specifies that the error occured is ORA-25228; it is thrown when the dequeuing timeout expires 
// or the end of the queue is reached without dequeuing a message.

int messageCount = 0;
while (messageCount < 2) {
  try {
    OracleQueueMessage msg = oracleDequeueQueue.Dequeue();
    messageCount++;
    if (msg != null && msg.StringPayload != null) {
      Console.WriteLine(msg.StringPayload);
    }
  }
  catch(OracleException ex) {
    if (ex.Code == 25228) {
      continue;
    }
    else
      throw ex;
  }
}

//Stop and destroy the queue and the queue table
oracleQueueAdmin.StopQueue();
oracleQueueAdmin.DropQueue();
oracleQueueTable.DropQueueTable();
oracleConnection.Close();

Visual Basic


Dim oracleConnection As New OracleConnection((
"User Id=system;Password=manager;Server=ora;")
oracleConnection.Open()

Dim oracleQueueTable As New OracleQueueTable("QUEUE_TABLE_MESSAGE", oracleConnection)

' Set sort order by priority for the queue. 
' The messages  with higher priority will reach the recipient first.
oracleQueueTable.Options.SortOrder = OracleQueueSortOrder.PriorityEnqueueTime

' Specify type of the messages in the queue. This time each message will be represented by just a string object. 
' We do so to simplify the unimportant 

oracleQueueTable.Options.PayloadTypeName = "RAW"

' The following operations are same as in the previous example.
oracleQueueTable.CreateQueueTable()
Dim oracleQueueAdmin As New OracleQueueAdmin("MESSAGE_QUEUE", _
   "QUEUE_TABLE_MESSAGE", oracleConnection)
oracleQueueAdmin.CreateQueue()
oracleQueueAdmin.StartQueue()
Dim oracleEnqueueQueue As New OracleQueue("MESSAGE_QUEUE", oracleConnection)

' Create and send the first message.
Dim message1 As New OracleQueueMessage
message1.StringPayload = "First message."
message1.MessageProperties.Priority = 1
oracleEnqueueQueue.Enqueue(message1)

' Create and send the second message. This message is assigned a higher priority value.
' The message will be consumed first, regardless of the fact that it was sent later.
Dim message2 As New OracleQueueMessage
message2.StringPayload = "Second message with high priority."
message2.MessageProperties.Priority = 7
oracleEnqueueQueue.Enqueue(message2)

' Create an object that receives the two messages.
Dim oracleDequeueQueue As New OracleQueue("MESSAGE_QUEUE", oracleConnection)
oracleDequeueQueue.DequeueOptions.WaitTimeout = 1

' Retrieve the messages in a loop. Once there are two messages received, quit.
' (ex.Code == 25228) specifies that the error occured is ORA-25228; 
' it is thrown when the dequeuing timeout expires or the end of the queue is reached 
' without dequeuing a message.

Dim messageCount As Integer = 0
Do While (messageCount < 2)
  Try
    Dim msg As OracleQueueMessage = oracleDequeueQueue.Dequeue()
    messageCount += 1
    if ((Not msg Is Nothing) AndAlso (Not msg.StringPayload Is Nothing)) Then
      Console.WriteLine(msg.StringPayload)
    End If
  Catch ex As OracleException
    If (ex.Code = 25228) Then
      Continue Do
    Else
      Throw ex
    End If
  End Try
Loop

oracleQueueAdmin.StopQueue()
oracleQueueAdmin.DropQueue()
oracleQueueTable.DropQueueTable()
oracleConnection.Close()

A Simple Example of Point-to-Multipoint Messaging

In the point-to-multipoint scheme one publisher generates messages for several subscribers. The publisher may specify who exactly will receive the message. By default messages do not expire, and they are deleted either when dequeued in Remove mode, or when the table is destroyed. You can assign a lifetime to every message when sending it.

The following example illustrates the point-to-multipoint scheme within a single application (though usually consumers are quite different applications). The example demonstrates another dequeue principle: every consumer fires an event when a new message is available.

Please note that AsyncNotification used in this example is supported only for OCI connections and is not supported for Direct connections.

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

void TestAQ() {
  OracleConnection oracleConnection = new OracleConnection(
     "User Id=system;Password=manager;Server=ora;");
  oracleConnection.Open();
  OracleQueueTable oracleQueueTable = new OracleQueueTable(
     "QUEUE_TABLE_MESSAGE", oracleConnection);

  // Set the table to serve multiple consumers.
  oracleQueueTable.Options.MultipleConsumers = true;
  oracleQueueTable.Options.PayloadTypeName = "RAW";
  oracleQueueTable.CreateQueueTable();
  OracleQueueAdmin oracleQueueAdmin = new OracleQueueAdmin(
     "MESSAGE_QUEUE", "QUEUE_TABLE_MESSAGE", oracleConnection);
  oracleQueueAdmin.CreateQueue();
  oracleQueueAdmin.StartQueue();

  // The OracleQueueAgent represents an AQ publisher or subscriber.
  // We register Bob and Scott as known subscribers so that the application is able to send messages to Bob or Scott personally.
  oracleQueueAdmin.AddSubscriber(new OracleQueueAgent("Bob"));
  oracleQueueAdmin.AddSubscriber(new OracleQueueAgent("Scott"));

  OracleQueue oracleEnqueueQueue = new OracleQueue("MESSAGE_QUEUE", oracleConnection);

  // This message will be delivered to Bob only.
  OracleQueueMessage message1 = new OracleQueueMessage();
  message1.StringPayload = "Message for Bob.";
  message1.MessageProperties.RecipientList.Add(new OracleQueueAgent("Bob"));

  // This message will be delivered to all recipients, as there is 
  // no specific member in the recipients list.
  OracleQueueMessage message2 = new OracleQueueMessage();
  message2.StringPayload = "Message for all.";

  // Create an OracleQueue instance that will receive general notifications 
  // and notifications for Scott.
  // The Navigation property specifies which message of the queue will be retrieved.
  OracleQueue oracleDequeueScott = new OracleQueue("MESSAGE_QUEUE", oracleConnection);
  oracleDequeueScott.DequeueOptions.Navigation = OracleQueueNavigation.FirstMessage;
  oracleDequeueScott.DequeueOptions.ConsumerName = "Scott";

  // Set up asynchronous notification using events.
  // The event just signals that a new message is available; 
  // messages are not dequeued implicitly.
  oracleDequeueScott.AsyncNotification = true;
  oracleDequeueScott.OnMessage += new OracleQueueMessageEventHandler(
        oracleDequeueScott_OnMessage);

  // Create an OracleQueue instance that will receive general 
  // notifications and notifications for Bob.
  OracleQueue oracleDequeueBob = new OracleQueue("MESSAGE_QUEUE", oracleConnection);
  oracleDequeueBob.DequeueOptions.Navigation = OracleQueueNavigation.FirstMessage;
  oracleDequeueBob.DequeueOptions.ConsumerName = "Bob";
  oracleDequeueBob.AsyncNotification = true;
  oracleDequeueBob.OnMessage += new OracleQueueMessageEventHandler(
        oracleDequeueBob_OnMessage);

  // Enqueue the messages
  oracleEnqueueQueue.Enqueue(message1);
  oracleEnqueueQueue.Enqueue(message2);

  // Wait a little so the events can fire.
  System.Threading.Thread.Sleep(5000);

  oracleQueueAdmin.StopQueue();
  oracleQueueAdmin.DropQueue();
  oracleQueueTable.DropQueueTable();

  oracleConnection.Close();
}


// A helper function that dequeues messages during an event
  string getMessageData(object sender, OracleQueueMessageEventArgs e) {
  OracleQueue oracleDequeue = sender as OracleQueue;
  OracleQueueDequeueOptions dequeueOptions = oracleDequeue.DequeueOptions;
  dequeueOptions.MessageId = e.MessageId;
  // Messages are not deleted from server once dequeued.
  dequeueOptions.DequeueMode = OracleQueueDequeueMode.Browse;
  OracleQueueMessage msg = oracleDequeue.Dequeue(dequeueOptions);
  return msg.StringPayload;
}


// Simple event handlers that write message text into console 
// when a message is received
void oracleDequeueBob_OnMessage(object sender, OracleQueueMessageEventArgs e) {
  string messageData = getMessageData(sender, e);
  Console.WriteLine("Bob receives a message: " + messageData);
}


void oracleDequeueScott_OnMessage(object sender, OracleQueueMessageEventArgs e) {
  string messageData = getMessageData(sender, e);
  Console.WriteLine("Scott receives a message: " + messageData);
}

Visual Basic


Private Sub TestAQ()
  Dim oracleConnection As New OracleConnection( _
       "User Id=system;Password=manager;Server=ora;")
  oracleConnection.Open()
  Dim oracleQueueTable As New OracleQueueTable( _
       "QUEUE_TABLE_MESSAGE", oracleConnection)

  ' Set the table to serve multiple consumers.
  oracleQueueTable.Options.MultipleConsumers = True
  oracleQueueTable.Options.PayloadTypeName = "RAW"
  oracleQueueTable.CreateQueueTable()
  Dim oracleQueueAdmin As New OracleQueueAdmin( _
        "MESSAGE_QUEUE", "QUEUE_TABLE_MESSAGE", oracleConnection)
  oracleQueueAdmin.CreateQueue()
  oracleQueueAdmin.StartQueue()

  ' The OracleQueueAgent represents an AQ publisher or subscriber.
  ' We register Bob and Scott as known subscribers so that the application is able to send messages to Bob or Scott personall
  oracleQueueAdmin.AddSubscriber(New OracleQueueAgent("Bob"))
  oracleQueueAdmin.AddSubscriber(New OracleQueueAgent("Scott"))

  Dim oracleEnqueueQueue As New OracleQueue("MESSAGE_QUEUE", oracleConnection)

  ' This message will be delivered to Bob only.
  Dim message1 As New OracleQueueMessage
  message1.StringPayload = "Message for Bob."
  message1.MessageProperties.RecipientList.Add(New OracleQueueAgent("Bob"))

  ' This message will be delivered to all recipients, as there is 
  ' no specific member in the recipients list.
  Dim message2 As New OracleQueueMessage
  message2.StringPayload = "Message for all."

  ' Create an OracleQueue instance that will receive general notifications 
  ' and notifications for Scott.
  ' The Navigation property specifies which message of the queue will be retrieved.
  Dim oracleDequeueScott As New OracleQueue("MESSAGE_QUEUE", oracleConnection)
  oracleDequeueScott.DequeueOptions.Navigation = OracleQueueNavigation.FirstMessage
  oracleDequeueScott.DequeueOptions.ConsumerName = "Scott"

  ' Set up asynchronous notification using events.
  ' The event just signals that a new message is available;   
  oracleDequeueScott.AsyncNotification = True
  AddHandler oracleDequeueScott.OnMessage, _
      New OracleQueueMessageEventHandler(AddressOf oracleDequeueScott_OnMessage)

  ' Create an OracleQueue instance that will receive general 
  ' notifications and notifications for Bob.
  Dim oracleDequeueBob As New OracleQueue("MESSAGE_QUEUE", oracleConnection)
  oracleDequeueBob.DequeueOptions.Navigation = OracleQueueNavigation.FirstMessage
  oracleDequeueBob.DequeueOptions.ConsumerName = "Bob"
  oracleDequeueBob.AsyncNotification = True
  AddHandler oracleDequeueBob.OnMessage, _
      New OracleQueueMessageEventHandler(AddressOf oracleDequeueBob_OnMessage)

  ' Enqueue the messages
  oracleEnqueueQueue.Enqueue(message1)
  oracleEnqueueQueue.Enqueue(message2)

  ' Wait a little so the events can fire.
  System.Threading.Thread.Sleep(5000)

  oracleQueueAdmin.StopQueue()
  oracleQueueAdmin.DropQueue()
  oracleQueueTable.DropQueueTable()

  oracleConnection.Close()
End Sub

' A helper function that dequeues messages during an event
Private Function getMessageData(ByVal sender As Object, _
    ByVal e As OracleQueueMessageEventArgs) As String
  Dim oracleDequeue As OracleQueue = TryCast(sender, OracleQueue)
  Dim dequeueOptions As OracleQueueDequeueOptions = oracleDequeue.DequeueOptions
  dequeueOptions.MessageId = e.MessageId
  ' Messages are not deleted from server once dequeued.
  dequeueOptions.DequeueMode = OracleQueueDequeueMode.Browse
  Dim msg As OracleQueueMessage = oracleDequeue.Dequeue(dequeueOptions)
  Return msg.StringPayload
End Function

' Simple event handlers that write message text into console 
' when a message is received
Private Sub oracleDequeueBob_OnMessage(ByVal sender As Object, _
    ByVal e As OracleQueueMessageEventArgs)
  Dim messageData As String = getMessageData(sender, e)
  Console.WriteLine(("Bob receives a message: " + messageData))
End Sub

Private Sub oracleDequeueScott_OnMessage(ByVal sender As Object, _
    ByVal e As OracleQueueMessageEventArgs)
  Dim messageData As String = getMessageData(sender, e)
  Console.WriteLine(("Scott receives a message: " + messageData))
End Sub

As seen in the three examples in this article, most administrative tasks can be accomplished by OracleQueueTable and OracleQueueAdmin classes. Note that administering queues often requires a rich set of privileges in the database.

AQ Basics

AQ components in dotConnect for Oracle

A Simple Example of Point-to-Point Messaging

An Extended Example of Point-to-Point Messaging

A Simple Example of Point-to-Multipoint Messaging

See Also