spring - JmsListener 没有收到来自 Activemq 的消息

Question

我的代码中有一个 Spring JmsListener。它接收并使用了 2 天的消息，但是在这 2 天之后它突然没有收到来自外部 activemq 的消息。但是，它的队列中有一些待处理的消息。当我重置 activemq 和消费者时，消费者收到大量消息。当消息挂起时，消费者连接到 activemq(代表 spring 执行器日志)。日志和配置显示 activemq 没有向消费者推送消息。我有另一个像这个消费者这样的服务，它从其他队列接收消息。第二个消费者也有同样的问题。我想知道为什么会出现这个问题，我该如何解决这个问题。这是我的配置和日志:

消费者:

应用程序.xml:

spring.jms.pub-sub-domain=false
spring.jms.template.delivery-mode=persistent
spring.activemq.broker-url=${BROKER_URL:failover:(tcp://activemq1:61616,tcp://activemq2:61616)?maxReconnectDelay=2500}
spring.activemq.user=${BROKER_USER:admin}
spring.activemq.password=${BROKER_PASSWORD:admin}

Jms配置:

@EnableJms
@Configuration
public class JmsConfiguration {
    @Bean
    public JmsListenerContainerFactory<?> myFactory(ActiveMQConnectionFactory connectionFactory,
                                                    DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setMaximumRedeliveries(0);
        connectionFactory.setRedeliveryPolicy(redeliveryPolicy);
        factory.setMessageConverter(messageConverter());
        configurer.configure(factory, connectionFactory);
        return factory;
    }

    private MessageConverter messageConverter() {
        MappingJackson2MessageConverter messageConverter = new MappingJackson2MessageConverter();
        messageConverter.setTargetType(MessageType.TEXT);
        messageConverter.setTypeIdPropertyName("_type");
        return messageConverter;
    }
}

听众:

@Component
public class MessageReceiver {


        @JmsListener(destination = Constant.OFFICE_REQUEST_QUEUE, containerFactory = "myFactory")
        public void receive(RequestMessage requestMessage,  @Headers Map<String,Object> headers) throws NonPersistenceServiceException {
            // do someting on received message
        }
    }

Activemq 配置:

activemq.xml:

<beans
  xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
  http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">

    <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="locations">
            <value>file:${activemq.conf}/credentials.properties</value>
        </property>
    </bean>

    <bean id="logQuery" class="io.fabric8.insight.log.log4j.Log4jLogQuery"
          lazy-init="false" scope="singleton"
          init-method="start" destroy-method="stop">
    </bean>

    <bean id="oracleDS" class="oracle.jdbc.pool.OracleDataSource" destroy-method="close">
          <property name="URL" value="jdbc:oracle:thin:@(DESCRIPTION= (SDU=32768)(ADDRESS=(PROTOCOL=TCP)(HOST= dms-db1-vip)(PORT=1521)) (ADDRESS=(PROTOCOL=TCP)(HOST= dms-db2-vip)(PORT=1521)) (LOAD_BALANCE=yes)(FAILOVER=ON)(CONNECT_DATA=(SERVER=DEDICATED)(SERVICE_NAME=orcl)(failover_mode=(type=select)(method=basic)(retries=5)(delay=1))))"/>
          <property name="user" value="ora_user"/>
          <property name="password" value="ora_pass"/>
    </bean>         

    <broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" >

        <plugins>
            <redeliveryPlugin fallbackToDeadLetter="true"
                              sendToDlqIfMaxRetriesExceeded="true">
                <redeliveryPolicyMap>
                    <redeliveryPolicyMap>

                        <defaultEntry>
                            <redeliveryPolicy 
                                              useExponentialBackOff="true"
                                              backOffMultiplier="2"
                                              maximumRedeliveryDelay="1200000"
                                              maximumRedeliveries="82"/> 
                        </defaultEntry>
                    </redeliveryPolicyMap>
                </redeliveryPolicyMap>
            </redeliveryPlugin>
        </plugins>

        <destinationPolicy>
            <policyMap>
              <policyEntries>
                <policyEntry queue=">">
                  <deadLetterStrategy>
                    <individualDeadLetterStrategy queuePrefix="DLQ." useQueueForQueueMessages="true"/>
                  </deadLetterStrategy>
                </policyEntry>
              </policyEntries>
            </policyMap>            
        </destinationPolicy>

        <managementContext>
            <managementContext createConnector="false"/>
        </managementContext>

        <persistenceAdapter>
            <jdbcPersistenceAdapter dataSource="#oracleDS" />
        </persistenceAdapter>

          <systemUsage>
            <systemUsage>
                <memoryUsage>
                    <memoryUsage percentOfJvmHeap="70" />
                </memoryUsage>
                <storeUsage>
                    <storeUsage limit="100 gb"/>
                </storeUsage>
                <tempUsage>
                    <tempUsage limit="50 gb"/>
                </tempUsage>
            </systemUsage>
        </systemUsage>

        <transportConnectors>
            <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
        </transportConnectors>

        <shutdownHooks>
            <bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
        </shutdownHooks>

    </broker>

    <import resource="jetty.xml"/>

</beans>

下面是consumer中actuator的日志:

{"log":"20:20:26.775 @@@ [ActiveMQ Task-1] INFO  o.a.activemq.transport.failover.FailoverTransport @@@  @@@  @@@\r\n","stream":"stdout"}
{"log":"                Successfully connected to tcp://activemq1:61616\r\n","stream":"stdout"}
{"log":"20:20:36.638 @@@ [ActiveMQ Task-1] INFO  o.a.activemq.transport.failover.FailoverTransport @@@  @@@  @@@\r\n","stream":"stdout"}
{"log":"                Successfully connected to tcp://activemq1:61616\r\n","stream":"stdout"}
{"log":"20:20:36.779 @@@ [ActiveMQ Task-1] INFO  o.a.activemq.transport.failover.FailoverTransport @@@  @@@  @@@\r\n","stream":"stdout"}
{"log":"                Successfully connected to tcp://activemq1:61616\r\n","stream":"stdout"}
{"log":"20:20:46.464 @@@ [ActiveMQ Task-1] INFO  o.a.activemq.transport.failover.FailoverTransport @@@  @@@  @@@\r\n","stream":"stdout"}
{"log":"                Successfully connected to tcp://activemq1:61616\r\n","stream":"stdout"}
{"log":"20:20:46.774 @@@ [ActiveMQ Task-1] INFO  o.a.activemq.transport.failover.FailoverTransport @@@  @@@  @@@\r\n","stream":"stdout"}

Activemq 错误日志:

2019-03-05 20:19:55,448 | WARN  | Transport Connection to: tcp://10.42.1.0:63493 failed: java.io.IOException: Frame size of 1 GB larger than max allowed 100 MB | org.apache.activemq.broker.TransportConnection.Transport | ActiveMQ Transport: tcp:///10.42.1.0:63493@61616
2019-03-05 20:20:05,776 | WARN  | Transport Connection to: tcp://10.42.1.0:63498 failed: java.io.EOFException | org.apache.activemq.broker.TransportConnection.Transport | ActiveMQ Transport: tcp:///10.42.1.0:63498@61616
2019-03-05 20:20:05,776 | WARN  | Transport Connection to: tcp://10.42.1.0:63497 failed: java.io.EOFException | org.apache.activemq.broker.TransportConnection.Transport | ActiveMQ Transport: tcp:///10.42.1.0:63497@61616

Answer 1

此类情况通常是以下两个问题之一的结果:

• 某些网络组件(路由器、防火墙)由于不活动而静默关闭套接字而不通知客户端或服务器，因此，除非启用心跳，否则连接已断开。

• 消费者线程以某种方式“卡在”用户代码中，因此它停止接收新消息；获取堆栈转储以查看容器线程正在做什么。