Общение с очередями через JMS

Задача: научиться получать и отправлять сообщения в очередь сообщений (Message Oriented Middleware).

С чем будем работать

Для работы с очередями в Java используются следующие классы:

• javax.jms.QueueConnectionFactory; 
• javax.jms.QueueConnection;
•  javax.jms.QueueSession;
• javax.jms.Queue;
• javax.jms.QueueSender;
• javax.jms.QueueReceiver;

QueueConnectionFactory
 Фабрика подключений. Используется для получения Connection.

QueueConnection
Представляет собой подключение, необходимое для создание одной или более сессий работы с очередью.

QueueSession
Используется для создания таких объектов как: QueueReceiver, QueueSender, QueueBrowser, и TemporaryQueue.

Queue
Объект, инкапсулирующий имя очереди специфичное для провайдера. По сути представляет собой очередь, чтение или запись из которой должна быть произведена.

QueueSender
 Используется для отправки сообщений.

QueueReceiver
Используется для получения сообщений.

Предварительные операции

Для отправки или получения сообщения необходимо создать объекты QueueReceiver или QueueSender соответственно. Последовательность вызовов объектов для этого следующая:

// 1) Получаем через JNDI менеджер подключений: 
QueueConnectionFactory qmanager = (QueueConnectionFactory)context.lookup("<Имя менеджера подключений>");

//  2) Получаем объект, представляющий очередь:
Queue queue = (Queue) context.lookup("<Имя очереди>"); 

// 3) Получаем и запускаем подключение: 
QueueConnection con =  qmanager.createQueueConnection();
con.start(); 

// 4) Начинаем сессию: 
QueueSession ses = con.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);

Особого внимания здесь требует метод createQueueSession. Первый его аргумент указывает, поддерживает ли подключение транзакции (если true, то значение второго аргумента игнорируется). Второй аргумент указывает кто из поставщика и клиента подтверждает, что сообщение получено. Приведенные в примере значения аргументов наиболее распространены.

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

Отправка сообщений

// 1) Перед отправкой сообщения необходимо это сообщение создать:
javax.jms.TextMessage message = session.createTextMessage("<Текст сообщения>");

// 2) Создаем отправителя (queue - очередь, в которую должно быть отправлено сообщение):
sender = session.createSender(queue);

// 3) Отправляем сообщение:
 sender.send(message);

// 4) Для снятия всех блокировок и сохранения сообщений, отправленных в сессии, выполняем:
session.commit();

Прием сообщений

Для приема сообщений существует два способа:

Способ первый

// 1) Создаем получателя
QueueReceiver receiver = ses.createReceiver(queue);

// 2) Получаем сообщение
javax.jms.Message message = receiver.receive();

Недостаток этого метода в том, что сообщения в очереди может и не быть на момент вызова метода receive(). Тогда поток будет остановлен до появления сообщения.

Способ второй

Заключается в том, чтобы реализовать интерфейс javax.jms.MessageListener и передать его получателю (receiver).

MessageListener listener = new MessageListener() {
 public void onMessage(Message message) {
  // Выполняются необходимые действия с сообщением
 }
};

receiver.setMessageListener(listener);

Интерфейс javax.jms.MessageListener предписывает реализовать всего один метод void onMessage(javax.jms.Message message). Как не трудно догадаться, этот метод вызывается получателем при получении сообщения (полученное сообщение передается в качестве аргумента метода), при этом выполнение потока не приостанавливается.

Напоследок

Иногда удобнее создавать объекты, связанные с очередями в обход jms с помощью специфических API, таких как API предоставляемые IBM MQ. Но это тема отдельного поста.