Как поставить сообщение JMS в Oracle AQ, используя Java

Question

У меня есть Oracle AQ с типом очереди SYS.AQ$_JMS_TEXT_MESSAGE. Я пытаюсь вставить текст в указанную очередь из Java-приложения.

Эквивалентный SQL-запрос

declare
 r_enqueue_options    DBMS_AQ.ENQUEUE_OPTIONS_T;
 r_message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
 v_message_handle     RAW(16);
 o_payload            SYS.AQ$_JMS_TEXT_MESSAGE;
begin
 o_payload := sys.aq$_jms_text_message.construct;
 o_payload.set_text(xmltype('<user>text</user>').getClobVal());
 sys.dbms_aq.enqueue (
   queue_name         => 'QUEUE_NAME',
   enqueue_options    => r_enqueue_options,
   message_properties => r_message_properties,
   payload            => o_payload,
   msgid              => v_message_handle
 );
 commit;
end;
/

Я понял это правильно, используя это руководство, но я застрял на

 o_payload := sys.aq$_jms_text_message.construct;
 o_payload.set_text(xmltype('<user>text</user>').getClobVal());

Руководство показывает, как поставить в очередь сообщение RAW, но мне нужно, чтобы оно было JMS, в противном случае тип данных не соответствует типу очереди.

Любая помощь будет оценена, потому что даже со всемогущим Google я не могу найти решение этой проблемы. Есть ли способ сделать это с помощью oracle.jdbc.aq классы, или я просто должен смириться с этим и использовать запрос SQL?

Answer 1

Просто скопируйте, вставьте этот код и попробуйте. (если это работает для вас) Затем внимательно изучите код и поймите.

Выполняя,

• Сначала раскомментируйте строку ' createQueue () ' в методе main.

после этого,

• Прокомментируйте его, раскомментируйте строку sendMessage() и попробуйте отправить сообщение.

Затем прокомментируйте / раскомментируйте каждую строку соответственно и попробуйте.

import java.util.ArrayList;
import java.util.Enumeration;
import java.util.List;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.Session;
import javax.jms.TextMessage;

import oracle.AQ.AQQueueTable;
import oracle.AQ.AQQueueTableProperty;
import oracle.jms.AQjmsDestination;
import oracle.jms.AQjmsDestinationProperty;
import oracle.jms.AQjmsFactory;
import oracle.jms.AQjmsSession;

public class OracleAQClient {

public static QueueConnection getConnection() {

    String hostname = "localhost";
    String oracle_sid = "xe";
    int portno = 1521;
    String userName = "jmsuser";
    String password = "jmsuser";
    String driver = "thin";
    QueueConnectionFactory QFac = null;
    QueueConnection QCon = null;
    try {
        // get connection factory , not going through JNDI here
        QFac = AQjmsFactory.getQueueConnectionFactory(hostname, oracle_sid, portno, driver);
        // create connection
        QCon = QFac.createQueueConnection(userName, password);
    } catch (Exception e) {
        e.printStackTrace();
    }
    return QCon;
}

public static void createQueue(String user, String qTable, String queueName) {
    try {
        /* Create Queue Tables */
        System.out.println("Creating Queue Table...");
        QueueConnection QCon = getConnection();
        Session session = QCon.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE);

        AQQueueTableProperty qt_prop;
        AQQueueTable q_table = null;
        AQjmsDestinationProperty dest_prop;
        Queue queue = null;
        qt_prop = new AQQueueTableProperty("SYS.AQ$_JMS_TEXT_MESSAGE");

        q_table = ((AQjmsSession) session).createQueueTable(user, qTable, qt_prop);

        System.out.println("Qtable created");
        dest_prop = new AQjmsDestinationProperty();
        /* create a queue */
        queue = ((AQjmsSession) session).createQueue(q_table, queueName, dest_prop);
        System.out.println("Queue created");
        /* start the queue */
        ((AQjmsDestination) queue).start(session, true, true);

    } catch (Exception e) {
        e.printStackTrace();
        return;
    }
}

public static void sendMessage(String user, String queueName,String message) {

    try {
        QueueConnection QCon = getConnection();
        Session session = QCon.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE);
        QCon.start();
        Queue queue = ((AQjmsSession) session).getQueue(user, queueName);
        MessageProducer producer = session.createProducer(queue);
        TextMessage tMsg = session.createTextMessage(message);

        //set properties to msg since axis2 needs this parameters to find the operation
        tMsg.setStringProperty("SOAPAction", "getQuote");
        producer.send(tMsg);
        System.out.println("Sent message = " + tMsg.getText());

        session.close();
        producer.close();
        QCon.close();

    } catch (JMSException e) {
        e.printStackTrace();
        return;
    }
}

public static void browseMessage(String user, String queueName) {
    Queue queue;
    try {
        QueueConnection QCon = getConnection();
        Session session = QCon.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE);

        QCon.start();
        queue = ((AQjmsSession) session).getQueue(user, queueName);
        QueueBrowser browser = session.createBrowser(queue);
        Enumeration enu = browser.getEnumeration();
        List list = new ArrayList();
        while (enu.hasMoreElements()) {
            TextMessage message = (TextMessage) enu.nextElement();
            list.add(message.getText());
        }
        for (int i = 0; i < list.size(); i++) {
            System.out.println("Browsed msg " + list.get(i));
        }
        browser.close();
        session.close();
        QCon.close();

    } catch (JMSException e) {
        e.printStackTrace();
    }

}

public static void consumeMessage(String user, String queueName) {
    Queue queue;
    try {
        QueueConnection QCon = getConnection();
        Session session = QCon.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE);
        QCon.start();
        queue = ((AQjmsSession) session).getQueue(user, queueName);
        MessageConsumer consumer = session.createConsumer(queue);
        TextMessage msg = (TextMessage) consumer.receive();
        System.out.println("MESSAGE RECEIVED " + msg.getText());

        consumer.close();
        session.close();
        QCon.close();
    } catch (JMSException e) {
        e.printStackTrace();
    }
}

public static void main(String args[]) {
    String userName = "jmsuser";
    String queue = "sample_aq";
    String qTable = "sample_aqtbl";
    //createQueue(userName, qTable, queue);
    //sendMessage(userName, queue,"<user>text</user>");
    //browseMessage(userName, queue);
    //consumeMessage(userName, queue);
}

}

Вам нужно будет скопировать эти jars/libs в ваш проект java из установочной директории oracle DB

1. ojdbc6.jar
2. jta.jar
3. jmscommon.jar
4. aqapi.jar

Авторы должны обратиться к Рате за эту статью [1]. Было немного вещей, которые нужно было изменить, я просто изменил их и предоставил код.

[1] http://wso2.com/library/tutorials/2011/11/configuring-wso2-esb-with-oracle-as-messaging-media/

Спасибо

Answer 2

Я добавлю некоторые лакомые кусочки в ответ @Chathura Kulasinghe.

Во-первых, в методе takeMessage, используя

Session.CLIENT_ACKNOWLEDGE

Параметр для создания объекта сеанса будет иметь эффект оставления сообщения, которое вы потребляете, в очереди. Если вы запустите эту программу много раз, вы увидите количество сообщений в таблице базы данных очереди. Чтобы удалить сообщение, вам необходимо "подтвердить" его, вызвав этот метод для объекта сообщения:

msg.acknowledge ();

Во-вторых, если вы хотите, чтобы сессия делала это для вас, просто измените режим подтверждения клиента на:

Session.AUTO_ACKNOWLEDGE

С этим параметром каждый раз, когда ваш customer.receive() - это вызов, автоматическое подтверждение и так далее, удаляемый из очереди.