我正在实施 Oracle Advanced Queue 并且对它完全陌生。我对此有一些疑问。下面是我的代码：
package com;
/* Set up main class from which we will call subsequent examples and handle
exceptions: */
import java.sql.*;
import oracle.AQ.*;
public class test_aqjava
{
public static void main(String args[])
{
AQSession aq_sess = null;
try
{
aq_sess = createSession(args);
createAqTables(aq_sess);
enqueueMsg(aq_sess);
// dequeueMsg(aq_sess);
aq_sess.close();
/* now run the test: */
// runTest(aq_sess);
}
catch (Exception ex)
{
System.out.println("Exception-1: " + ex);
ex.printStackTrace();
}
}
public static AQSession createSession(String args[])
{
Connection db_conn;
AQSession aq_sess = null;
try
{
Class.forName("oracle.jdbc.driver.OracleDriver");
/* your actual hostname, port number, and SID will
vary from what follows. Here we use 'dlsun736,' '5521,'
and 'test,' respectively: */
db_conn =
DriverManager.getConnection(
"jdbc:oracle:thin:@hostname.com:1521:sid",
"USER", "USER");
System.out.println("JDBC Connection opened ");
db_conn.setAutoCommit(false);
/* Load the Oracle8i AQ driver: */
Class.forName("oracle.AQ.AQOracleDriver");
/* Creating an AQ Session: */
aq_sess = AQDriverManager.createAQSession(db_conn);
System.out.println("Successfully created AQSession ");
}
catch (Exception ex)
{
System.out.println("Exception: " + ex);
ex.printStackTrace();
}
return aq_sess;
}
public static void createAqTables(AQSession aq_sess) throws AQException
{
AQQueueTableProperty qtable_prop;
AQQueueProperty queue_prop;
AQQueueTable q_table;
AQQueue queue;
/* Creating a AQQueueTableProperty object (payload type - RAW): */
qtable_prop = new AQQueueTableProperty("RAW");
/* Creating a queue table called aq_table1 in aqjava schema: */
q_table = aq_sess.createQueueTable ("USER", "aq_table1", qtable_prop);
System.out.println("Successfully created aq_table1 in aqjava schema");
/* Creating a new AQQueueProperty object */
queue_prop = new AQQueueProperty();
/* Creating a queue called aq_queue1 in aq_table1: */
queue = aq_sess.createQueue (q_table, "aq_queue1", queue_prop);
System.out.println("Successfully created aq_queue1 in aq_table1");
/* Enable enqueue/dequeue on this queue: */
queue.start();
System.out.println("Successful start queue");
}
public static void enqueueMsg(AQSession aq_sess) throws AQException
{
AQQueueTable q_table;
AQQueue queue;
AQMessage message;
AQRawPayload raw_payload;
AQEnqueueOption enq_option;
String test_data = "new message";
byte[] b_array;
Connection db_conn;
db_conn = ((AQOracleSession)aq_sess).getDBConnection();
/* Get a handle to queue table - aq_table4 in aqjava schema: */
q_table = aq_sess.getQueueTable ("USER", "aq_table1");
System.out.println("Successful getQueueTable");
/* Get a handle to a queue - aq_queue4 in aquser schema: */
queue = aq_sess.getQueue ("USER", "aq_queue1");
System.out.println("Successful getQueue");
/* Creating a message to contain raw payload: */
message = queue.createMessage();
/* Get handle to the AQRawPayload object and populate it with raw data: */
b_array = test_data.getBytes();
raw_payload = message.getRawPayload();
raw_payload.setStream(b_array, b_array.length);
/* Creating a AQEnqueueOption object with default options: */
enq_option = new AQEnqueueOption();
/* Enqueue the message: */
queue.enqueue(enq_option, message);
try {
db_conn.commit();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
public static void dequeueMsg(AQSession aq_sess) throws AQException
{
AQQueueTable q_table;
AQQueue queue;
AQMessage message;
AQRawPayload raw_payload;
AQDequeueOption deq_option;
byte[] b_array;
Connection db_conn;
db_conn = ((AQOracleSession)aq_sess).getDBConnection();
/* Get a handle to queue table - aq_table4 in aqjava schema: */
q_table = aq_sess.getQueueTable ("USER", "aq_table1");
System.out.println("Successful getQueueTable");
/* Get a handle to a queue - aq_queue4 in aquser schema: */
queue = aq_sess.getQueue ("USER", "aq_queue1");
System.out.println("Successful getQueue");
/* Creating a AQDequeueOption object with default options: */
deq_option = new AQDequeueOption();
/* Enqueue the message: */
message = queue.dequeue(deq_option);
raw_payload = message.getRawPayload();
b_array= raw_payload.getBytes();
String msg = new String(b_array);
System.out.println("Dequeue Msg "+msg);
try {
db_conn.commit();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
我创建了一个队列表和一个队列。消息已从队列中写入和读取。
Q1。我可以在同一个队列中再写一条消息并从中读取吗？如果是，我们该怎么做？因为我尝试将消息写入同一个队列但不能
Q2。如何将上面的代码转换为发布订阅？如何通过多次阅读相同的消息来测试它？
任何帮助表示赞赏。