Весна Интеграция poller зависает или останавливается через некоторое время

Question

У меня есть приложение, которое читает сообщение JMS, поступающее из HornetQ.

<int-jms:message-driven-channel-adapter 
      id="myJmsChannelAdapter" 
      channel="myJmsChannel" 
      connection-factory="myCredentialsConnectionFactory" 
      destination-name="${my.jms.topic}" 
      pub-sub-domain="true" 
      subscription-durable="${my.jms.subscription.durable}" 
      client-id="${my.jms.client.id}" 
      durable-subscription-name="${my.jms.subscription.name}" 
      /> 

Когда сообщение приезжает, я отношусь к ним с помощью Poller

<!-- Send Jms message one by one to creation --> 
<int:bridge input-channel="myJmsChannel" output-channel="jmsMessageCreateInputChannel"> 
    <int:poller max-messages-per-poll="1" fixed-delay="100"> 
     <int:transactional transaction-manager="transactionManager" propagation="REQUIRES_NEW"/> 
    </int:poller> 
</int:bridge> 

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

<chain id="jmsMessageCreate" input-channel="jmsMessageCreateInputChannel"> 
    <!-- extract the operation from the message itself --> 
    <int-xml:xpath-header-enricher> 
     <int-xml:header name="operation" xpath-expression="name(./node())" overwrite="true"/> 
    </int-xml:xpath-header-enricher> 

    <service-activator expression="@jmsMessageService.createNewJmsMessage(payload, headers['operation'])"/> 
    <!-- have a JPA entity in the payload --> 
    <gateway request-channel="persistEntityChannel"/> 
    <logging-channel-adapter logger-name="JmsLogger" expression="'JMS message with operation ['+headers['operation']+'] created in DB'"/> 
</chain> 

Моя проблема иногда, без видимых причин и без сообщения об ошибке в журналах, мои сообщения больше не обрабатываются. Цепь id jmsMessageCreate больше не вызывается, как если бы poller перестала работать. Если сервер перезагружен, он снова работает хорошо, но я потерял сообщения JMS, которые были отправлены между моментом остановки опроса и моментом перезапуска сервера.

Я действительно не знаю, где искать или что делать, чтобы предотвратить это. Может быть, изменение в конфиге?

Заранее за вашу помощь.

Answer 1

Скорее всего, поток опроса «застрял» где-то в вашем коде. Используйте jstack <pid>, чтобы взять дамп потока, чтобы посмотреть, что делает поток. Вы можете использовать jps, чтобы найти pid.

Чтобы избежать потери сообщений, вы должны установить acknowledge="transacted" на адаптере, управляемом сообщениями, - это было изменено на значение по умолчанию true в версии 4.2.

Вы не должны использовать QueueChannels и pollers в этой среде - управление потоками управляется адаптером, управляемым сообщениями. С каналом очереди, даже с транзакционными сеансами, контейнер будет зафиксирован, как только сообщение будет помещено в очередь.