Oracle Advanced Queuing (AQ) provides database-integrated message queuing functionality. It is built on top of Oracle Streams and optimizes 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, HTTP, and HTTPS. Because Oracle AQ is implemented in database tables, all operational benefits of high availability, scalability, and reliability are also applicable to queue data. This chapter provides information about the Java interface to Oracle AQ.
This chapters covers the following topics:
Oracle Database 11g Release 1 (11.1) provides a fast Java interface to AQ by introducing a new Java package,oracle.jdbc.aq. This package contains the following:
Classes
AQDequeueOptions
Specifies the options available for the dequeue operation
AQEnqueueOptions
Specifies the options available for the enqueue operation
AQFactory
Is a factory class for AQ
AQNotificationEvent
Is created whenever a new message is enqueued in a queue for which you have registered your interest
Interfaces
AQAgent
Used to represent and identify a user of the queue or a producer or consumer of the message
AQMessage
Represents a message that is enqueued or dequeued
AQMessageProperties
Contains message properties such as Correlation, Sender, Delay and Expiration, Recipients, and Priority and Ordering
AQNotificationListener
Is a listener interface for receiving AQ notification events
AQNotificationRegistration
Represents your interest in being notified when a new message is enqueued in a particular queue
These classes and interfaces enable you to access an existing queue, create messages, and enqueue and dequeue messages.
Note:
Oracle JDBC drivers donot provide any API to create a queue. Queues must be created through theDBMS_AQADM PL/SQL package.See Also:
For more information about the APIs, refer to the Javadoc.A JDBC application can do the following:
Register interest in notifications in the AQ namespace and receive notifications when an enqueue occurs
Register interest in subscriptions to database events and receive notifications when the events are triggered
Registered clients are notified asynchronously when events are triggered or on an explicit AQ enqueue (or a new message is enqueued in a queue for which you have registered your interest). Clients do not need to be connected to a database.
Example 25-1 illustrates the use of the new JDBC AQ interface which is new feature of the JDBC Thin driver in release 11.1. It shows how to enqueue and dequeue from aRAW type single-consumer and multiple-consumer queue. It also shows how AQ asynchronous notification works. In this example, theSCOTT user is connecting to the database. Therefore, in the database, you must grant the following privileges to the user:
GRANT EXECUTE ON DBMS_AQ to SCOTT;GRANT EXECUTE ON DBMS_AQADM to SCOTT;GRANT AQ_ADMINISTRATOR_ROLE TO SCOTT;GRANT ADMINISTER DATABASE TRIGGER TO SCOTT;
Example 25-1 AQ Asynchronous Event Notification Example
import java.sql.*;import java.util.Properties;import oracle.jdbc.*;import oracle.jdbc.aq.*;public class DemoAQRawQueue{ static final String USERNAME= "scott"; static final String PASSWORD= "tiger"; static final String URL = "jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS=(PROTOCOL=tcp)(HOST=oracleserver.mydomain.com) (PORT=1521))"+"(CONNECT_DATA=(SERVICE_NAME=mydatabaseinstance)))"; public static final void main(String[] argv) { DemoAQRawQueue demo = new DemoAQRawQueue(); try { demo.run(); }catch(SQLException ex) { ex.printStackTrace(); } } void run() throws SQLException { OracleConnection connection = connect(); connection.setAutoCommit(false); cleanup(connection); setup(connection); // run the demo for single consumer queue: demoSingleConsumerQueue(connection); // run the demo for multi consumer queue: demoMultipleConsumerQueue(connection); connection.close(); } /** * Single consumer queue demo: * This method enqueues a dummy message into the RAW_SINGLE_QUEUE queue and dequeues it. * Then it registers for AQ notification on this same queue and enqueues a * message again. The AQ listener will be notified that a new message has arrived * in the queue and it will dequeue it. */ void demoSingleConsumerQueue(OracleConnection connection) throws SQLException { System.out.println("\n ============= Start single consumer queue demo ============= \n"); String queueType = "RAW"; String queueName = USERNAME+".RAW_SINGLE_QUEUE"; enqueueDummyMessage(connection,queueName,null); dequeue(connection,queueName,queueType,null); AQNotificationRegistration reg = registerForAQEvents( connection,queueName); DemoAQRawQueueListener single_li = new DemoAQRawQueueListener( queueName,queueType); reg.addListener(single_li); enqueueDummyMessage(connection,queueName,null); connection.commit(); while(single_li.getEventsCount() < 1) { try { Thread.currentThread().sleep(1000); } catch (InterruptedException e) { } } single_li.closeConnection(); connection.unregisterAQNotification(reg); } /** * Multi consumer queue demo: * This method first registers for AQ notification upon the agent "BLUE". It * then enqueues a message for "RED" and "BLUE" */ void demoMultipleConsumerQueue(OracleConnection connection) throws SQLException { System.out.println("\n ============= Start multi consumer queue demo ============= \n"); String queueType = "RAW"; String queueName = USERNAME+".RAW_MULTIPLE_QUEUE"; AQNotificationRegistration reg = registerForAQEvents( connection,queueName+":BLUE"); DemoAQRawQueueListener multi_li = new DemoAQRawQueueListener( queueName,queueType); reg.addListener(multi_li); AQAgent[] recipients = new AQAgent[2]; recipients[0] = AQFactory.createAQAgent(); recipients[0].setName("BLUE"); recipients[1] = AQFactory.createAQAgent(); recipients[1].setName("RED"); enqueueDummyMessage(connection,queueName,recipients); connection.commit(); while(multi_li.getEventsCount() < 1) { try { Thread.currentThread().sleep(1000); } catch (InterruptedException e) { } } dequeue(connection,queueName,queueType,"RED"); multi_li.closeConnection(); connection.unregisterAQNotification(reg); } /** * This method enqueues a dummy message in the queue specified by its name. */ public void enqueueDummyMessage(OracleConnection conn, String queueName, AQAgent[] recipients) throws SQLException { System.out.println("----------- Enqueue start ------------"); // First create the message properties: AQMessageProperties msgprop = AQFactory.createAQMessageProperties(); msgprop.setCorrelation("mycorrelation"); msgprop.setExceptionQueue("MY_EXCEPTION_QUEUE"); // Specify an agent as the sender: AQAgent ag = AQFactory.createAQAgent(); ag.setName("MY_SENDER_AGENT_NAME"); ag.setAddress("MY_SENDER_AGENT_ADDRESS"); msgprop.setSender(ag); // handle multi consumer case: if(recipients != null) msgprop.setRecipientList(recipients); System.out.println(msgprop.toString()); // Create the actual AQMessage instance: AQMessage mesg = AQFactory.createAQMessage(msgprop); // and add a payload: byte[] rawPayload = new byte[500]; for(int i=0;i<rawPayload.length;i++) rawPayload[i] = 'b'; mesg.setPayload(new oracle.sql.RAW(rawPayload)); // We want to retrieve the message id after enqueue: AQEnqueueOptions opt = new AQEnqueueOptions(); opt.setRetrieveMessageId(true); // execute the actual enqueue operation: conn.enqueue(queueName,opt,mesg); byte[] mesgId = mesg.getMessageId(); if(mesgId != null) { String mesgIdStr = byteBufferToHexString(mesgId,20); System.out.println("Message ID from enqueue call: "+mesgIdStr); } System.out.println("----------- Enqueue done ------------"); } /** * This methods dequeues the next available message from the queue specified * by "queueName". */ public void dequeue(OracleConnection conn, String queueName, String queueType, String consumerName ) throws SQLException { System.out.println("----------- Dequeue start ------------"); AQDequeueOptions deqopt = new AQDequeueOptions(); deqopt.setRetrieveMessageId(true); if(consumerName != null) deqopt.setConsumerName(consumerName); // dequeue operation: AQMessage msg = conn.dequeue(queueName,deqopt,queueType); // print out the message that has been dequeued: byte[] payload = msg.getPayload(); byte[] msgId = msg.getMessageId(); if(msgId != null) { String mesgIdStr = byteBufferToHexString(msgId,20); System.out.println("ID of message dequeued = "+mesgIdStr); } AQMessageProperties msgProp = msg.getMessageProperties(); System.out.println(msgProp.toString()); String payloadStr = new String(payload,0,10); System.out.println("payload.length="+payload.length+", value="+payloadStr); System.out.println("----------- Dequeue done ------------"); } public AQNotificationRegistration registerForAQEvents( OracleConnection conn, String queueName) throws SQLException { Properties globalOptions = new Properties(); String[] queueNameArr = new String[1]; queueNameArr[0] = queueName; Properties[] opt = new Properties[1]; opt[0] = new Properties(); opt[0].setProperty(OracleConnection.NTF_AQ_PAYLOAD,"true"); AQNotificationRegistration[] regArr = conn.registerAQNotification(queueNameArr,opt,globalOptions); AQNotificationRegistration reg = regArr[0]; return reg; } /** * */ private void setup(Connection conn) throws SQLException { doUpdateDatabase(conn, "BEGIN "+ "DBMS_AQADM.CREATE_QUEUE_TABLE( "+ " QUEUE_TABLE => '"+USERNAME+".RAW_SINGLE_QUEUE_TABLE', "+ " QUEUE_PAYLOAD_TYPE => 'RAW', "+ " COMPATIBLE => '10.0'); "+ "END; "); doUpdateDatabase(conn, "BEGIN "+ "DBMS_AQADM.CREATE_QUEUE( "+ " QUEUE_NAME => '"+USERNAME+".RAW_SINGLE_QUEUE', "+ " QUEUE_TABLE => '"+USERNAME+".RAW_SINGLE_QUEUE_TABLE'); "+ "END; "); doUpdateDatabase(conn, "BEGIN "+ " DBMS_AQADM.START_QUEUE('"+USERNAME+".RAW_SINGLE_QUEUE'); "+ "END; "); // create a multi consumer RAW queue: doUpdateDatabase(conn, "BEGIN "+ "DBMS_AQADM.CREATE_QUEUE_TABLE( "+ " QUEUE_TABLE => '"+USERNAME+".RAW_MULTIPLE_QUEUE_TABLE', "+ " QUEUE_PAYLOAD_TYPE => 'RAW', "+ " MULTIPLE_CONSUMERS => TRUE, "+ " COMPATIBLE => '10.0'); "+ "END; "); doUpdateDatabase(conn, "BEGIN "+ "DBMS_AQADM.CREATE_QUEUE( "+ " QUEUE_NAME => '"+USERNAME+".RAW_MULTIPLE_QUEUE', "+ " QUEUE_TABLE => '"+USERNAME+".RAW_MULTIPLE_QUEUE_TABLE'); "+ "END; "); doUpdateDatabase(conn, "BEGIN "+ " DBMS_AQADM.START_QUEUE('"+USERNAME+".RAW_MULTIPLE_QUEUE'); "+ "END; "); } void cleanup( Connection conn) { tryUpdateDatabase(conn, "BEGIN "+ " DBMS_AQADM.STOP_QUEUE('"+USERNAME+".RAW_SINGLE_QUEUE'); "+ "END; "); tryUpdateDatabase(conn, "BEGIN "+ " DBMS_AQADM.DROP_QUEUE_TABLE( "+ " QUEUE_TABLE => '"+USERNAME+".RAW_SINGLE_QUEUE_TABLE', "+ " FORCE => TRUE); "+ "END; "); tryUpdateDatabase(conn, "BEGIN "+ " DBMS_AQADM.STOP_QUEUE('"+USERNAME+".RAW_MULTIPLE_QUEUE'); "+ "END; "); tryUpdateDatabase(conn, "BEGIN "+ " DBMS_AQADM.DROP_QUEUE_TABLE( "+ " QUEUE_TABLE => '"+USERNAME+".RAW_MULTIPLE_QUEUE_TABLE', "+ " FORCE => TRUE); "+ "END; "); } /** * Creates a connection the database. */ OracleConnection connect() throws SQLException { OracleDriver dr = new OracleDriver(); Properties prop = new Properties(); prop.setProperty("user",DemoAQRawQueue.USERNAME); prop.setProperty("password",DemoAQRawQueue.PASSWORD); return (OracleConnection)dr.connect(DemoAQRawQueue.URL,prop); } /** * Utility method: executes a DML query but doesn't throw any exception if * an error occurs. */ private void tryUpdateDatabase(Connection conn,String sql) { Statement stmt = null; try { stmt = conn.createStatement(); stmt.executeUpdate(sql); }catch(SQLException sqlex) { System.out.println("Exception ("+sqlex.getMessage()+") while trying to execute \""+sql+"\""); } finally { if(stmt != null) { try { stmt.close(); }catch(SQLException exx){exx.printStackTrace();} } } } /** * Utility method: executes a DML query and throws an exception if * an error occurs. */ private void doUpdateDatabase(Connection conn,String sql) throws SQLException { Statement stmt = null; try { stmt = conn.createStatement(); stmt.executeUpdate(sql); } finally { if(stmt != null) { try { stmt.close(); }catch(SQLException exx){} } } } static final String byteBufferToHexString(byte[] buffer, int maxNbOfBytes) { if(buffer == null) return null; int offset = 0; boolean isFirst = true; StringBuffer sb = new StringBuffer(); while(offset < buffer.length && offset < maxNbOfBytes) { if(!isFirst) sb.append(' '); else isFirst = false; String hexrep = Integer.toHexString((int)buffer[offset]&0xFF); if(hexrep.length() == 1) hexrep = "0"+hexrep; sb.append(hexrep); offset++; } String ret= sb.toString(); return ret; }} class DemoAQRawQueueListener implements AQNotificationListener{ OracleConnection conn; String queueName; String typeName; int eventsCount = 0; public DemoAQRawQueueListener(String _queueName, String _typeName) throws SQLException { queueName = _queueName; typeName = _typeName; conn = (OracleConnection)DriverManager.getConnection (DemoAQRawQueue.URL, DemoAQRawQueue.USERNAME, DemoAQRawQueue.PASSWORD); } public void onAQNotification(AQNotificationEvent e) { System.out.println("\n----------- DemoAQRawQueueListener: got an event ----------- "); System.out.println(e.toString()); System.out.println("------------------------------------------------------"); System.out.println("----------- DemoAQRawQueueListener: Dequeue start -----------"); try { AQDequeueOptions deqopt = new AQDequeueOptions(); deqopt.setRetrieveMessageId(true); if(e.getConsumerName() != null) deqopt.setConsumerName(e.getConsumerName()); if((e.getMessageProperties()).getDeliveryMode() == AQMessageProperties.MESSAGE_BUFFERED) { deqopt.setDeliveryMode(AQDequeueOptions.DEQUEUE_BUFFERED); deqopt.setVisibility(AQDequeueOptions.DEQUEUE_IMMEDIATE); } AQMessage msg = conn.dequeue(queueName,deqopt,typeName); byte[] msgId = msg.getMessageId(); if(msgId != null) { String mesgIdStr = DemoAQRawQueue.byteBufferToHexString(msgId,20); System.out.println("ID of message dequeued = "+mesgIdStr); } System.out.println(msg.getMessageProperties().toString()); byte[] payload = msg.getPayload(); if(typeName.equals("RAW")) { String payloadStr = new String(payload,0,10); System.out.println("payload.length="+payload.length+", value="+payloadStr); } System.out.println("----------- DemoAQRawQueueListener: Dequeue done -----------"); } catch(SQLException sqlex) { System.out.println(sqlex.getMessage()); } eventsCount++; } public int getEventsCount() { return eventsCount; } public void closeConnection() throws SQLException { conn.close(); }}Before you enqueue a message, you must create the message. An instance of a class implementing theAQMessage interface represents an AQ message. An AQ message contains properties (metadata) and a payload (data). Perform the following to create an AQ message:
Create an instance ofAQMessageProperties.
Set the property attributes.
Create the AQ message using theAQMessageProperties object.
Set the payload.
The properties of the AQ message are represented by an instance of theAQMessageProperties interface. You can set or get the following message properties:
Dequeue Attempts Count: Specifies the number of attempts that have been made to dequeue the message. This property cannot be set.
Correlation: Is an identifier supplied by the producer of the message at the time of enqueuing the message.
Delay: Is the number of seconds for which the message is in theWAITING state. After the specified delay, the message is in theREADY state and available for dequeuing. Dequeuing a message by using the message ID (msgid) overrides the delay specification.
Note:
Delay is not supported with buffered messaging.Delivery Mode: Specifies whether the message is a buffered message or a persistent message. This property cannot be set.
Enqueue Time: Specifies the time at which the message was enqueued. This value is determined by the system and cannot be set by the user.
Exception Queue: Specifies the name of the queue into which the message is moved if it cannot be processed successfully. Messages are moved in two cases:
The number of unsuccessful dequeue attempts has exceededmax_retries.
The message has expired.
Expiration: Is the number of seconds during which the message is available for dequeuing, starting from when the message reaches theREADY state. If the message is not dequeued before it expires, then it is moved to the exception queue in theEXPIRED state.
Message State: Specifies the state of the message at the time of dequeuing the message. This property cannot be set.
Previous Queue Message ID: Is the ID of the message in the last queue that generated the current message. When a message is propagated from one queue to another, this attribute identifies the ID of the queue from which it was last propagated. This property cannot be set.
Priority: Specifies the priority of the message. It can be any integer including negative integers; the smaller the value, the higher the priority.
Recipient list: Is a list ofAQAgent objects that represent the recipients. The default recipients are the queue subscribers. This parameter is valid only for multiple-consumer queues.
Sender: Is an identifier specified by the producer at the time of enqueuing the message. It is an instance ofAQAgent.
Transaction group: Specifies the transaction group of the message for transaction-grouped queues. It is set after a successful call to thedequeueArray method.
The following code snippet illustrates how to create anAQMessageProperties object and create an AQ message using it:
...// Create the message properties objectAQMessageProperties msgprop = AQFactory.createAQMessageProperties();// Set some properties (optional)msgprop.setCorrelation("mycorrelation");msgprop.setExceptionQueue("MY_EXCEPTION_QUEUE");msgprop.setExpiration(0);msgprop.setPriority(1);AQAgent ag = AQFactory.createAQAgent();ag.setName("MY_SENDER_AGENT_NAME");ag.setAddress("MY_SENDER_AGENT_ADDRESS");msgprop.setSender(ag);msgprop.setRecipientList(recipients);// Create the messageAQMessage mesg = AQFactory.createAQMessage(msgprop);...Depending on the type of the queue, the payload of the AQ message can be specified using thesetPayload method of theAQMessage interface. The following code snippet illustrates how to set the payload:
...byte[] rawPayload = new byte[500];for(int i=0;i<rawPayload.length;i++) rawPayload[i] = 'b';mesg.setPayload(new oracle.sql.RAW(rawPayload));...
You can retrieve the payload of an AQ message using thegetPayload method or the appropriategetXXXPayload method. These methods are defined in theAQMessage interface.
After you create a message and set the message properties and payload, you can enqueue the message using theenqueue method of theOracleConnection interface. Before you enqueue the message, you can specify some enqueue options. TheAQEnqueueOptions class enables you to specify the following enqueue options:
Delivery mode: Specifies the delivery mode. Delivery mode can be set to either persistent (ENQUEUE_PERSISTENT) or buffered (ENQUEUE_BUFFERED).
Retrieve Message ID: Specifies whether or not the message ID has to be retrieved from the server when the message has been enqueued. By default, the message ID is not retrieved.
Transformation: Specifies a transformation that will be applied before enqueuing the message. The return type of the transformation function must match the type of the queue.
Note:
Transformations must be created in PL/SQL usingDBMS_TRANSFORM.CREATE_TRANSFORMATION(...).Visibility: Specifies the transactional behavior of the enqueue request. The default value for this option isENQUEUE_ON_COMMIT. It indicates that the enqueue operation is part of the current transaction.ENQUEUE_IMMEDIATE indicates that the enqueue operation is an autonomous transaction, which commits at the end of the operation. For buffered messaging, you must useENQUEUE_IMMEDIATE.
The following code snippet illustrates how to set the enqueue options and enqueue the message:
...// Set the enqueue optionsAQEnqueueOptions opt = new AQEnqueueOptions();opt.setRetrieveMessageId(true);// Enqueue the message// conn is an instance of OracleConnection// queueName is a String// mesg is an instance of AQMessageconn.enqueue(queueName, opt, mesg);...
Enqueued messages can be dequeued using thedequeue method of theOracleConnection interface. Before you dequeue a message you must set the dequeue options. TheAQDequeueOptions class enables you to specify the following dequeue options:
Condition: Specifies a conditional expression based on the message properties, the message data properties, and PL/SQL functions. A dequeue condition is specified as aBoolean expression using syntax similar to theWHERE clause of a SQL query.
Consumer name: If specified, only the messages matching the consumer name are accessed.
Note:
If the queue is a single-consumer queue, donot set this option.Correlation: Specifies a correlation criterion (or search criterion) for the dequeue operation.
Delivery Filter: Specifies the type of message to be dequeued. You dequeue buffered messages only (DEQUEUE_BUFFERED) or persistent messages only (DEQUEUE_PERSISTENT), which is the default, or both (DEQUEUE_PERSISTENT_OR_BUFFERED).
Dequeue Message ID: Specifies the message identifier of the message to be dequeued. This can be used to dequeue a unique message whose ID is known.
Dequeue mode: Specifies the locking behavior associated with the dequeue operation. It can take one of the following values:
DequeueMode.BROWSE: Message is dequeued without acquiring any lock.
DequeueMode.LOCKED: Message is dequeued with a write lock that lasts for the duration of the transaction.
DequeueMode.REMOVE: (default) Message is dequeued and deleted. The message can be retained in the queue based on the retention properties.
DequeueMode.REMOVE_NO_DATA: Message is marked as updated or deleted.
Maximum Buffer Length: Specifies the maximum number of bytes that will be allocated when dequeuing a message from aRAW queue. The default maximum isDEFAULT_MAX_PAYLOAD_LENGTH but it can be changed to any other nonzero value. If the buffer is not large enough to contain the entire message, then the exceeding bytes will be silently ignored.
Navigation: Specifies the position of the message that will be retrieved. It can take one of the following values:
NavigationOption.FIRST_MESSAGE: The first available message matching the search criteria is dequeued.
NavigationOption.NEXT_MESSAGE: (default) The next available message matching the search criteria is dequeued. If the previous message belongs to a message group, then the next available message matching the search criteria in the message group is dequeued.
NavigationOption.NEXT_TRANSACTION: Messages in the current transaction group are skipped, and the first message of the next transaction group is dequeued. This setting can be usedonly if message grouping is enabled for the queue.
Retrieve Message ID: Specifies whether or not the message identifier of the dequeued message needs to be retrieved. By default, it is not retrieved.
Transformation: Specifies a transformation that will be applied after dequeuing the message. The source type of the transformation must match the type of the queue.
Note:
Transformations must be created in PL/SQL usingDBMS_TRANSFORM.CREATE_TRANSFORMATION(...).Visibility: Specifies whether or not the message is dequeued as part of the current transaction. It can take one of the following values:
VisibilityOption.ON_COMMIT: (default) The dequeue operation is part of the current transaction.
VisibilityOption.IMMEDIATE: The dequeue operation is an autonomous transaction that commits at the end of the operation.
Note:
The Visibility option is ignored in theDequeueMode.BROWSE dequeue mode. If the delivery filter isDEQUEUE_BUFFERED orDEQUEUE_PERSISTENT_OR_BUFFERED, then this optionmust be set toVisibilityOption.IMMEDIATE.Wait: Specifies the wait time for the dequeue operation, if none of the messages matches the search criteria. The default value isDEQUEUE_WAIT_FOREVER indicating that the operation waits forever. If set toDEQUEUE_NO_WAIT, then the operation does not wait. If a number is specified, then the dequeue operation waits for the specified number of seconds.
Note:
If you useDEQUEUE_WAIT_FOREVER, then the dequeue operation will not return until a message that matches the search criterion is available in the queue. However, you can interrupt the dequeue operation by calling thecancel method on theOracleConnection object.The following code snippet illustrates how to set the dequeue options and dequeue the message:
...// Set the dequeue optionsAQDequeueOptions deqopt = new AQDequeueOptions();deqopt.setRetrieveMessageId(true);deqopt.setConsumerName(consumerName);// Dequeue the message// conn is an OracleConnection object// queueName is a String identifying the queue// queueType is a String specifying the type of the queue, such as RAWAQMessage msg = conn.dequeue(queueName,deqopt,queueType);
This section provides a few examples that illustrate how to enqueue and dequeue messages.
Example 25-2 illustrates how to enqueue a message, andExample 25-3 illustrates how to dequeue a message.
Example 25-2 Enqueuing a Single Message
This example illustrates how to obtain access to a queue, create a message, and enqueue it.
// Get access to the queue// conn is an instance of oracle.jdbc.OracleConnectionoracle.jdbc.aq.AQQueue queue = conn.getAQQueue("SCOTT.MY_QUEUE","RAW");// Create the message properties objectoracle.jdbc.aq.AQMessageProperties msgprop =oracle.jdbc.aq.AQMessageProperties.createAQMessageProperties();// Set some properties (optional)msgprop.setPriority(1);msgprop.setExceptionQueue("EXCEPTION_QUEUE");msgprop.setExpiration(0);AQAgent agent = AQAgent.createAQAgent("AGENTNAME");ag2.setAddress("AGENTADDRESS");msgprop.setSender(agent);// Create the messageAQMessage mesg = AQMessage.createAQMessage(msgprop);// Set the payload// where buffer contains the RAW payload:mesg.setRawPayload(new AQRawPayload(buffer));// Set the enqueue optionsAQEnqueueOptions options = new AQEnqueueOptions();//Set the enqueue options using the setXXX methods of AQEnqueueOptions.// Enqueue the messagequeue.enqueue(options, mesg);Example 25-3 Dequeuing a Single Message
This example illustrates how to obtain access to a queue, set the dequeue options, and dequeue the message.
// Get access to the queue// conn is an instance of oracle.jdbc.OracleConnectionoracle.jdbc.aq.AQQueue queue = conn.getAQQueue ("SCOTT.MY_QUEUE","RAW");// Set the dequeue optionsAQDequeueOptions options = new AQDequeueOptions();options.setDeliveryMode(AQDequeueOptions.BUFFERED);// Dequeue the messageAQMessage mesg = queue.dequeue(options);