Дополнительные функции JMS

В этом разделе объясняется, как использовать функции JMS API для достижения необходимого уровня надёжности и производительности. Многие люди используют JMS в своих приложениях, потому что их приложения не переносят потерянных или дублирующихся сообщений и требуют, чтобы каждое сообщение принималось один и только один раз. JMS API обеспечивает им эту функциональность.

Самый надёжный способ создания сообщения — отправить сообщение PERSISTENT и сделать это в транзакции.

Сообщения JMS по умолчанию PERSISTENT. Сообщения PERSISTENT не теряются при сбоях провайдера JMS. Для получения дополнительной информации см. Указание персистентности сообщения.

Транзакции позволяют отправлять или получать несколько сообщений в атомарной операции. В платформе Java EE они также позволяют комбинировать отправку и получение сообщений со считыванием и записью базы данных в рамках одной транзакции. Транзакция — это атомарное действие, которое может объединять ряд операций, таких как отправка и получение сообщений, так что либо все операции выполняются успешно, либо все завершаются неудачно. Подробнее см. Использование локальных транзакций JMS.

Самый надёжный способ получения сообщений — делать это в транзакции, неважно из очереди или из долговременной подписки на тему. Для получения дополнительной информации см. Создание долговременных подписок, Создание вре́менных пунктов назначения и Использование локальных транзакций JMS.

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

Также могут быть указаны различные уровни контроля над подтверждением сообщения. Смотрите Управление подтверждением сообщения.

Другие функции могут предоставить полезные возможности, не связанные с надёжностью. Например, могут быть созданы вре́менные пункты назначения, которые будут действовать только на время создавшего их соединения. Смотрите Создание вре́менных пунктов назначения для получения подробной информации.

В следующих разделах описываются эти функции применительно к клиентским приложениям или клиентам Java SE. Некоторые функции работают по-разному в веб-контейнере Java EE или контейнере EJB. В этих случаях различия отмечены здесь и подробно объяснены в Использование JMS API в приложениях Java EE.

Управление подтверждением сообщения

Пока сообщение JMS не будет подтверждено, оно не считается успешно доставленным. Успешная доставка сообщения обычно состоит из трёх этапов.

Клиент получает сообщение.
Клиент обрабатывает сообщение.
Сообщение подтверждается. Подтверждение инициируется либо провайдером JMS, либо клиентом, в зависимости от режима подтверждения сессии.

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

В транзакции JTA (в веб-контейнере Java EE или в контейнере EJB) сообщение подтверждается, когда транзакция фиксируется.

В нетранзакционных сессиях время и способ подтверждения сообщения зависят от значения, которое может быть указано в аргументе метода createContext. Возможные значения аргумента следующие.

JMSContext.AUTO_ACKNOWLEDGE: этот параметр используется по умолчанию для клиентских приложений и клиентов Java SE. JMSContext автоматически подтверждает получение клиентом сообщения, либо когда клиент успешно возвратил вызов receive, либо когда вызов MessageListener для обработки сообщения возвращается успешно.

Синхронный приём в JMSContext, настроенный на использование автоматического подтверждения, является единственным исключением из правила, согласно которому получение сообщений является трёхэтапным процессом, как описано ранее. В этом случае получение и подтверждение происходит за один шаг, после чего следует обработка сообщения.

JMSContext.CLIENT_ACKNOWLEDGE: клиент подтверждает сообщение, вызывая метод acknowledge сообщения. В этом режиме подтверждение происходит на уровне сессии: подтверждение полученного сообщения автоматически подтверждает получение всех сообщений, которые были получены его сессией. Например, если потребитель сообщения получает десять сообщений, а затем подтверждает пятое доставленное сообщение, все десять сообщений подтверждаются.

Примечание:

В платформе Java EE параметр JMSContext.CLIENT_ACKNOWLEDGE можно использовать только в клиентском приложении, но не в веб-компоненте или Enterprise-бине.

JMSContext.DUPS_OK_ACKNOWLEDGE: эта опция указывает JMSContext отложенно (lazy) подтверждать доставку сообщений. Это может привести к дублированию в доставке сообщений в случае сбоя JMS-провайдера, поэтому его должны использовать только потребители, терпимые к дубликатам сообщений. (Если провайдер JMS повторно доставляет сообщение, он должен установить значение заголовка сообщения JMSRedelivered в true.) Этот параметр может уменьшить накладные расходы сессии, сводя к минимуму работу, которую выполняет сессия для предотвращения дублирования.

Если сообщения были получены из очереди, но не подтверждены, когда JMSContext закрыт, провайдер JMS сохраняет и повторно доставляет их при следующем обращении потребителя к очереди. Поставщик также сохраняет неподтверждённые сообщения, если приложение закрывает JMSContext, который получал сообщения из длительной подписки. (См. Создание долговременных подписок.) Неподтверждённые сообщения, полученные из подписки, не являющейся долговременной, будут потеряны при закрытии JMSContext.

Если используется очередь или долговременная подписка, можно использовать метод JMSContext.recover, чтобы остановить нетранзакционный JMSContext и перезапустить его с его первым неподтверждённым сообщением. Фактически, серия доставленных сообщений JMSContext сбрасывается в точку после последнего подтверждённого сообщения. Сообщения, которые он теперь доставляет, могут отличаться от первоначально доставленных, если срок действия сообщений истёк или были получены сообщения с более высоким приоритетом. Для потребителя с подпиской, не являющейся долговременной, поставщик может отбрасывать неподтверждённые сообщения при вызове метода JMSContext.recover.

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

Указание параметров для отправки сообщений

Вы можете установить ряд параметров при отправке сообщения. Эти параметры позволяют выполнять задачи, описанные в следующих разделах:

Указание персистентности сообщений — укажите, что сообщения являются персистентными, то есть они не должны быть потеряны в случае сбоя провайдера.
Установка приоритета сообщений — установите сообщениям приоритет, который может повлиять на порядок доставки сообщений.
Установка срока действия сообщения — укажите срок действия сообщений, чтобы устаревшие сообщения не доставлялись.
Указание задержки доставки — укажите задержку доставки сообщений, чтобы они не доставлялись до истечения указанного времени.
Использование цепочки методов JMSProducer — цепочка методов позволяет вам указать более одной из этих опций при создании производителя и вызове метода send.

Указание персистентности сообщения

JMS API поддерживает два режима доставки, определяющие, будут ли сообщения потеряны в случае сбоя провайдера JMS. Эти режимы доставки являются полями интерфейса DeliveryMode.

Режим доставки по умолчанию, PERSISTENT, инструктирует провайдера JMS проявить особую осторожность, чтобы гарантировать, что сообщение не будет потеряно при передаче в случае сбоя провайдера JMS. Сообщение, отправленное в этом режиме доставки, записывается в стабильное хранилище при его отправке.
Режим доставки NON_PERSISTENT не требует от провайдера JMS сохранять сообщение или иным образом гарантировать, что оно не будет потеряно в случае сбоя провайдера.

Для указания режима доставки используйте метод setDeliveryMode интерфейса JMSProducer, чтобы установить режим доставки для всех сообщений, отправленных этим производителем.

Для установки режима доставки при создании производителя и отправке сообщения может использоваться цепочка методов. Следующий вызов создаёт производителя с режимом доставки NON_PERSISTENT и использует его для отправки сообщения:

context.createProducer()
       .setDeliveryMode(DeliveryMode.NON_PERSISTENT).send(dest, msg);

Если не указан режим доставки, по умолчанию используется значение PERSISTENT. Режим доставки NON_PERSISTENT может повысить производительность и снизить накладные расходы на хранение, но должен использоваться только в том случае, если приложение терпимо к потере сообщений.

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

Используйте приоритеты сообщений, чтобы инструктировать JMS-провайдера сначала доставлять срочные сообщения. Используйте метод setPriority интерфейса JMSProducer для установки приоритета всем сообщений, отправленных этим производителем.

Используйте цепочку методов для установки приоритета при создании производителя и отправке сообщения. Например, следующий вызов устанавливает приоритет уровня 7 для производителя и затем отправляет сообщение:

context.createProducer().setPriority(7).send(dest, msg);

Десять уровней приоритета варьируются от 0 (самый низкий) до 9 (самый высокий). Если не указано, по умолчанию используется приоритет уровня 4. Провайдер JMS пытается доставлять сообщения с более высоким приоритетом раньше, чем сообщения с более низким приоритетом, но не обязан доставлять сообщения в точном порядке приоритетов.

Установка срока действия сообщения

По умолчанию сообщение никогда не истекает. Если сообщение через некоторое время устареет, ему может быть установлен срок действия. Для установки срока действия по умолчанию для всех сообщений, отправленных этим производителем, используется метод setTimeToLive интерфейса JMSProducer.

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

Для установки времени действия при создании производителя и отправке сообщения может быть использована цепочка методов. Например, следующий вызов устанавливает пять минут для производителя, а затем отправляет сообщение:

context.createProducer().setTimeToLive(300000).send(dest, msg);

Если указанное значение timeToLive равно 0, срок действия сообщения никогда не истечёт.

Когда сообщение отправлено, указанный timeToLive добавляется к текущему времени, чтобы указать время действия. Любое сообщение, не доставленное до истечения указанного срока, уничтожается. Уничтожение устаревших сообщений экономит ресурсы хранения и вычислительные ресурсы.

Указание задержки доставки

Можно указать период времени, который должен пройти после отправки сообщения, прежде чем провайдер JMS доставит сообщение. Используйте метод setDeliveryDelay интерфейса JMSProducer, чтобы установить задержку доставки для всех сообщений, отправленных этим производителем.

Для установки задержки доставки при создании производителя и отправке сообщения может использоваться цепочка методов. Например, следующий вызов устанавливает задержку доставки в 3 секунды для производителя, а затем отправляет сообщение:

context.createProducer().setDeliveryDelay(3000).send(dest, msg);

Использование цепочки методов JMSProducer

Set-методы в интерфейсе JMSProducer возвращают объекты JMSProducer, поэтому может быть использована цепочка методов для создания производителя, установки нескольких свойств и отправки сообщения. Например, следующие вызовы связанных методов создают производителя, устанавливают пользовательское свойство, срок действия, режим доставки и приоритет сообщения, а затем отправляют сообщение в очередь:

context.createProducer()
        .setProperty("MyProperty", "MyValue")
        .setTimeToLive(10000)
        .setDeliveryMode(NON_PERSISTENT)
        .setPriority(2)
        .send(queue, body);

Также могут быть вызваны методы JMSProducer для установки свойств сообщения, а затем отправлено сообщение отдельным вызовом метода send. Свойства сообщения можно установить непосредственно в сообщении.

Создание вре́менных пунктов назначения

Как правило, назначения JMS (очереди и темы) создаются административно, а не программно. JMS-провайдер включает в себя инструмент для создания и удаления пунктов назначения, и пункты назначения обычно являются долговременными.

JMS API также позволяет создавать пункты назначения (объекты TemporaryQueue и TemporaryTopic), действующие только на время соединения, в котором они созданы. Эти пункты назначения создаются динамически методами JMSContext.createTemporaryQueue и JMSContext.createTemporaryTopic, как в следующем примере:

TemporaryTopic replyTopic = context.createTemporaryTopic();

Единственные потребители сообщений, которые могут получать из врéменного пункта назначения, — это пользователи, созданные тем же соединением, которое создало пункт назначения. Производитель может отправить любое сообщение во врéменный пункт назначения. Если закрывается соединение, которому принадлежит врéменный пункт назначения, пункт назначения также закрывается и его содержимое теряется.

Вре́менные пункты назначения могут использоваться для реализации простого механизма запроса/ответа. При создании врéменного пункта назначения и указании его в качестве значения поля заголовка сообщения JMSReplyTo при отправке сообщения получатель сообщения может использовать значение поля JMSReplyTo в качестве пункта назначения, на которое он отправляет ответ. Потребитель также может ссылаться на исходный запрос, задав в поле заголовка JMSCorrelationID ответного сообщения значение поля заголовка JMSMessageID запроса. Например, метод onMessage может создать JMSContext, чтобы он мог отправить ответ на полученное сообщение. Он может использовать такой код:

replyMsg = context.createTextMessage("Consumer processed message: "
        + msg.getText());
replyMsg.setJMSCorrelationID(msg.getJMSMessageID());
context.createProducer().send((Topic) msg.getJMSReplyTo(), replyMsg);

Пример см. в разделе Использование сущности для объединения сообщений из двух MDB.

Использование локальных транзакций JMS

Транзакция выполняет ряд операций атомарно. Если какая-либо из операций завершается неудачно, транзакцию можно откатить, а операции повторить с самого начала. Если все операции выполнены успешно, транзакция может быть зафиксирована.

В клиентском приложении или клиенте Java SE можно использовать локальные транзакции для группировки отправки и получения сообщений. Метод JMSContext.commit используется для совершения транзакции. Можно отправить несколько сообщений в транзакции, и сообщения не будут добавлены в очередь или тему, пока транзакция не будет зафиксирована. При получении нескольких сообщений в транзакции они не будут подтверждены до тех пор, пока транзакция не будет зафиксирована.

Метод JMSContext.rollback может использоваться для отката транзакции. Откат транзакции означает, что все созданные сообщения уничтожаются, а все использованные сообщения восстанавливаются и доставляются, если срок их действия не истёк (см. Установка срока действия сообщения).

Транзакционная сессия всегда участвует в транзакции. Чтобы создать транзакционную сессию, вызовите метод createContext следующим образом:

JMSContext context =
        connectionFactory.createContext(JMSContext.SESSION_TRANSACTED);

Как только вызывается метод commit или rollback, одна транзакция заканчивается и начинается другая. Закрытие транзакции отменяет текущую транзакцию, включая все ожидающие отправки и получения.

В приложении, работающем в веб-контейнере Java EE или контейнере EJB, нельзя использовать локальные транзакции. Вместо этого используются транзакции JTA, описанные в Использование JMS API в приложениях Java EE.

Несколько отправок и получений могут быть объединены в одной локальной транзакции JMS, если они все выполняются с использованием одного и того же JMSContext.

Не используйте одну транзакцию с механизмом запроса/ответа, в котором отправляется сообщение, а затем получается ответ на это сообщение. Если попытаться это сделать, программа зависнет, потому что отправка не может быть выполнена до тех пор, пока транзакция не будет зафиксирована. Следующий фрагмент кода иллюстрирует проблему:

// Не делайте этого!
outMsg.setJMSReplyTo(replyQueue);
context.createProducer().send(outQueue, outMsg);
consumer = context.createConsumer(replyQueue);
inMsg = consumer.receive();
context.commit();

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

Отправка и получение сообщения не могут быть частью одной транзакции. Причина заключается в том, что транзакции происходят между клиентами и провайдером JMS, который находится между отправкой и получением сообщения. Рисунок 48-8 иллюстрирует это взаимодействие.

Рис. 48-8. Использование локальных транзакций JMS

Диаграмма локальных транзакций, показывающая отдельные транзакции для отправки и получения сообщения

Отправка одного или нескольких сообщений одному или нескольким получателям клиентом 1 может сформировать одну транзакцию, поскольку она формирует единый набор взаимодействий с провайдером JMS с использованием одного JMSContext. Аналогично, получение одного или нескольких сообщений от одного или нескольких пунктов назначения клиентом 2 также формирует одну транзакцию с использованием одного JMSContext. Но поскольку оба клиента не имеют прямого взаимодействия и используют два разных объекта JMSContext, единая транзакция для них невозможна.

Другой способ объяснить это заключается в том, что транзакция — это контракт между клиентом и провайдером JMS, определяющий, отправлено ли сообщение в пункт назначения или получено ли сообщение из пункта назначения. Это не контракт между отправляющим и получающим клиентами.

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

При создании JMSContext можно указать, будет ли он транзакционным, передавая аргумент JMSContext.SESSION_TRANSACTED методу createContext. Например:

try (JMSContext context = connectionFactory.createContext(
        JMSContext.SESSION_TRANSACTED);) {
    ...

Методы commit и rollback для локальных транзакций связаны с сессией, которая лежит в основе JMSContext. Операции над несколькими очередями или темами или комбинациями очередей и тем могут быть объединены в одной транзакции, если для выполнения операций используется одна и та же сессия. Например, можно использовать тот же JMSContext, чтобы получить сообщение из очереди и отправить сообщение в тему в той же транзакции.

Пример в Использование локальных транзакций показывает, как использовать локальные транзакции JMS.

Асинхронная отправка сообщений

Обычно при отправке персистентного сообщения метод send блокируется до тех пор, пока провайдер JMS не подтвердит, что сообщение было успешно отправлено. Механизм асинхронной отправки позволяет приложению отправить сообщение и продолжить работу в ожидании информации о завершении отправки.

Эта функция в настоящее время доступна только в клиентских приложениях и клиентах Java SE.

Асинхронная отправка сообщения включает предоставление объекта Callback-вызова. Укажите CompletionListener с помощью метода onCompletion. Например, следующий код создаёт объект CompletionListener с именем SendListener. Затем он вызывает метод setAsync для указания, что отправка от этого производителя должна быть асинхронной и использован указанный слушатель:

CompletionListener listener = new SendListener();
context.createProducer().setAsync(listener).send(dest, message);

Класс CompletionListener должен реализовывать два метода: onCompletion и onException. Метод onCompletion вызывается в случае успешной отправки, а метод onException вызывается в случае сбоя. Простая реализация этих методов может выглядеть так:

@Override
public void onCompletion(Message message) {
    System.out.println("onCompletion method: Send has completed.");
}

@Override
public void onException(Message message, Exception e) {
    System.out.println("onException method: send failed: " + e.toString());
    System.out.println("Unsent message is: \n" + message);
}