Правильный подход для инициализации асинхронного JMS-слушателя и его бесконечной работы - PullRequest
Новая
       19

Правильный подход для инициализации асинхронного JMS-слушателя и его бесконечной работы

0 голосов
/ 19 ноября 2018

Я использую следующее приложение, чтобы подключиться к моей базе данных Oracle, зарегистрировать прослушиватель очереди и ждать сообщений в очереди: 

package sample;

import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSession;
import javax.jms.Session;

import oracle.jms.AQjmsFactory;
import oracle.jms.AQjmsSession;

public class MyThread extends Thread {

    private static final String QUEUE_NAME = "MY_QUEUE";
    private static final String QUEUE_USER = "myuser";
    private static final String QUEUE_PW = "mypassword";
    private boolean running;

    public MyThread() {
        this.running = true;
    }

    public static void main(String[] args) {
        MyThread mt = new MyThread();
        mt.start();
    }

    private QueueConnection getQueueConnection() throws JMSException {
        QueueConnectionFactory QFac = AQjmsFactory.getQueueConnectionFactory("xxx.xxx.xxx.xxx", "orcl", 1521, "thin");
        QueueConnection QCon = QFac.createQueueConnection(QUEUE_USER, QUEUE_PW);
        return QCon;
    }

    @Override
    public void interrupt() {
        this.running = false;
        super.interrupt();
    }

    @Override
    public void run() {
        try {
            QueueConnection queueConnection = getQueueConnection();
            QueueSession queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = ((AQjmsSession) queueSession).getQueue(QUEUE_USER, QUEUE_NAME);

            while (running) {
                System.out.println("Starting...");

                queueConnection.start();
                MessageConsumer mq = ((AQjmsSession) queueSession).createReceiver(queue);
                MyListener listener = new MyListener();
                mq.setMessageListener(listener);

                System.out.println("... Done, now sleep a bit and redo");

                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            System.out.println("Closing Application");
            queueSession.close();
            queueConnection.close();

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

Как только сообщение помещено в очередь, функция onMessage добавит сообщение в текстовый файл: 

package sample;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;

import javax.jms.Message;
import javax.jms.MessageListener;

public class MyListener implements MessageListener{

    @Override
    public void onMessage(Message arg0) {
        try {
            PrintWriter out = new PrintWriter(new BufferedWriter(new FileWriter("C:/temp/output/messages.txt", true)));
            out.println("New Message arrived");
            out.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

Во время выполнения мой вывод на консоль выглядит так:

Starting ...
... Готово, теперь поспать немного повторить
Запуск ...
... Готово, теперь поспать немного повторить
Запуск ...
... Готово, теперь поспать немного повторить
Запуск ...
... Готово, теперь поспать немного повторить

Если там, где есть какие-либо очереди, файл будет содержать новые сообщения.
Итак: я могу удалить события из очереди, и событие onMessage будет вызвано этим кодом.

Теперь на мой вопрос : я почти уверен, что ждать 5 секунд, чтобы снова зарегистрировать прослушиватель (и вызвать queueConnection.start () для приема вызовов onMessage) - неправильный подход. Но если я не сделаю этого, не будет никаких событий onMessage (Файл остается пустым).

Каков правильный подход, чтобы начать бесконечное прослушивание очереди без фиксированного вызова Thread.sleep () и без необходимости заново регистрировать прослушиватель, даже если не было никаких событий?

Дополнительная информация

База данных: Oracle 11g2
Java Runtime: 1.6
Зависимости Maven:

  • oracle-jdbc (11.2.0.4.0)
  • xdb (1,0)
  • Акапи (1,3)
  • jmscommon (1.3.1_02)

1 Ответ

0 голосов
/ 19 ноября 2018

Нет причин запускать поток для создания потребителя JMS и установки его прослушивателя сообщений.Весь смысл слушателя сообщения JMS состоит в том, чтобы получать сообщение асинхронно (функциональность, которую вы, по-видимому, пытаетесь продублировать по некоторым причинам).

Вам просто нужно создать получателя JMS и настроить прослушиватель сообщений, а затем убедиться, что получатель не закрыт.В зависимости от того, как написано приложение, иногда необходимо иметь цикл while, чтобы убедиться, что программа не завершает работу и, следовательно, закрывает пользователя.Ваша тема этого не делает.Это позволяет потребителю выпасть из области видимости после ожидания сообщений в течение 5 секунд, что означает, что оно будет собирать мусор, и я ожидаю, что для большинства реализаций JMS это означает, что оно будет закрыто.Хотя это может быть и хуже.Не закрывая явно клиента и просто выпуская его из области видимости, вы можете утекать потребителей, что в конечном итоге приведет к сбою вашего брокера сообщений.Это не только неаккуратное программирование, но потенциально проблематично для других пользователей, пытающихся использовать сообщения.

...