Персистентное сообщение в теме после перезапуска сервера

Question

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

В основном я запускаю ActiveMQ, используя REST Client. Я вызываю производителя для отправки сообщения в 50 раз, чтобы 50 сообщений попадали в очередь. На потребительском конце я применил таймер отключения 5 секунд, чтобы каждое сообщение потреблялось с регулярным интервалом 5 секунд. Затем между ними я остановил ActiveMQ. Между тем, некоторые сообщения потребляются клиентом, говорят, что 15 из 50 были израсходованы. Затем, если я перезапущу ActiveMQ, я ожидал, что тема останется в ожидании 35 сообщений, но я не могу видеть это в консоли администратора под вкладкой тем.

Вот мой конфигурационный файл:

<?xml version="1.0" encoding="UTF-8"?> 
<beans xmlns="http://www.springframework.org/schema/beans" 
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
     xmlns:context="http://www.springframework.org/schema/context" 
     xmlns:int="http://www.springframework.org/schema/integration" 
     xmlns:int-jms="http://www.springframework.org/schema/integration/jms" 
     xmlns:oxm="http://www.springframework.org/schema/oxm" 
     xmlns:int-jme="http://www.springframework.org/schema/integration" 
     xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd 
       http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd 
       http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd 
       http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd 
       http://www.springframework.org/schema/oxm http://www.springframework.org/schema/oxm/spring-oxm-3.0.xsd"> 


    <!-- Component scan to find all Spring components --> 
    <context:component-scan base-package="com.geekcap.springintegrationexample" /> 

    <bean class="org.springframework.web.servlet.mvc.annotation.AnnotationMethodHandlerAdapter"> 
     <property name="order" value="1" /> 
     <property name="messageConverters"> 
      <list> 
       <!-- Default converters --> 
       <bean class="org.springframework.http.converter.StringHttpMessageConverter"/> 
       <bean class="org.springframework.http.converter.FormHttpMessageConverter"/> 
       <bean class="org.springframework.http.converter.ByteArrayHttpMessageConverter" /> 
       <bean class="org.springframework.http.converter.xml.SourceHttpMessageConverter"/> 
       <bean class="org.springframework.http.converter.BufferedImageHttpMessageConverter"/> 
       <bean class="org.springframework.http.converter.json.MappingJackson2HttpMessageConverter" /> 
      </list> 
     </property> 
    </bean> 

    <!-- Define a channel to communicate out to a JMS Destination --> 
    <int:channel id="topicChannel"/> 

    <!-- Define the ActiveMQ connection factory --> 
    <bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory"> 
     <property name="brokerURL" value="tcp://localhost:61616"/> 
    </bean> 

    <!-- 
     Define an adaptor that route topicChannel messages to the myTopic topic; the outbound-channel-adapter 
     automagically fines the configured connectionFactory bean (by naming convention 
     --> 
    <int-jms:outbound-channel-adapter channel="topicChannel" 
             destination-name="topic.myTopic" 
             pub-sub-domain="true" /> 

    <!-- Create a channel for a listener that will consume messages--> 
    <int:channel id="listenerChannel" /> 

    <int-jms:message-driven-channel-adapter id="messageDrivenAdapter" 
              channel="getPayloadChannel" 
              destination-name="topic.myTopic" 
              pub-sub-domain="true" /> 

    <int:service-activator input-channel="listenerChannel" ref="messageListenerImpl" method="processMessage" /> 

    <int:channel id="getPayloadChannel" /> 

    <int:service-activator input-channel="getPayloadChannel" output-channel="listenerChannel" ref="retrievePayloadServiceImpl" method="getPayload" /> 

</beans> 

Я также читал, что режим по умолчанию является постоянным. Но в моем случае это не работает.

EDIT:

Согласно полученному ответу Гари Рассел после добавления атрибутов

• подписки долговечного = "истинный"
• -подписки имя долговечно = "mySubscription"

in <int-jms:message-driven-channel-adapter> Я столкнулся с вопросами, связанными с XML

• cvc-complex-type.3.2.2: Атрибут «подписка-долговечность» не может появляться в элементе «int-jms: message-driven-channel-adapter».

• cvc-complex-type.3.2.2: Атрибут «durable-subscription-name» не может появляться в элементе «int-jms: message-driven-channel-adapter».

Пожалуйста, помогите

Answer 1

Вот как темы работы, по умолчанию, прочитайте спецификацию JMS.

Темы публикуются/подписаны; только получатели сообщения получают сообщение.

Если вы публикуете 5, запустите потребителя, опубликуйте еще 5; он получит только второй 5.

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

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

Чтобы настроить это с помощью интеграции с пружиной, установите subscription-durable на адаптере канала с сообщением и дайте ему уникальный subscription-name.