ActiveMQ 代理错误“目标试图恢复的 JMS 消息侦听器调用程序的设置失败。原因：消费者已关闭”

Question

我们在环境中观察到 ActiveMQ UI 中的消费者被删除。我们的流量非常低，并且观察到最初我们有 3 个消费者，并且在间隔几个小时后将每个消费者都删除了，一旦我们重新启动消费者，它会再次刷新连接几个小时，我们在日志中看不到任何错误，除了：

Setup of JMS message listener invoker failed for destination 'queue-1' - trying to recover. Cause: The Consumer is closed

我使用 AWS ActiveMQ 代理，但在代理日志中看不到任何错误。

我们按照推荐使用PoolConnectionFactory 和ActiveMQConnectionFactory 为我们的消费者创建连接池。我们正在使用 ActiveMQ 5.15

@Bean
public PooledConnectionFactory pooledConnectionFactory() {
    ActiveMQConnectionFactory activeMQConnectionFactory =
            new ActiveMQConnectionFactory();
    activeMQConnectionFactory.setBrokerURL(brokerUrl);
    activeMQConnectionFactory.setUserName(username);
    activeMQConnectionFactory.setPassword(password);
    activeMQConnectionFactory.setTrustAllPackages(true);
    ActiveMQPrefetchPolicy activeMQPrefetchPolicy = new ActiveMQPrefetchPolicy();
    activeMQPrefetchPolicy.setQueuePrefetch(100);
    //activeMQPrefetchPolicy.setQueuePrefetch();
    activeMQConnectionFactory.setPrefetchPolicy(activeMQPrefetchPolicy);
    PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(activeMQConnectionFactory);
    pooledConnectionFactory.setMaxConnections(poolSize);
    return pooledConnectionFactory;
}

@Bean
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
    DefaultJmsListenerContainerFactory factory =
            new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(pooledConnectionFactory());
    factory.setMessageConverter(jacksonJmsMessageConverter());
    factory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
    factory.setConcurrency("1-1");
    factory.setErrorHandler(ActiveMQErrorHandler());
    return factory;
}

@Bean
public JmsTemplate jmsTemplate() {
    JmsTemplate jmsTemplate = new JmsTemplate(pooledConnectionFactory());
    jmsTemplate.setMessageConverter(jacksonJmsMessageConverter());
    return jmsTemplate;
}

@Bean
public Queue queue() {
    return new ActiveMQQueue(queueName);
}

@Bean
public ErrorHandler ActiveMQErrorHandler() {
    return t -> {
        LOGGER.error("JMS_LISTENER_ERROR");
    };
}

Answer 1

鉴于提供的信息，听起来好像连接正在断开并且客户端没有报告，或者远程正在关闭消费者，池可能不会注意到，直到执行某些用户操作。

这是使用 JMS 池的问题之一，即池无法完全了解客户端的情况，因此检查池中的连接可能会导致获取由于没有冒泡到池层的 IO 中断，陈旧且不再活动的连接。解决此问题的一种方法是使用 ActiveMQ 客户端故障转移传输，以允许它在连接断开时自动重新连接到代理。

您可以尝试的另一个选项是使用PooledJMS JMS 连接池库，它完成了一些额外的工作以尝试更快地验证失败的连接和/或关闭的资源，并将其与创建使用故障转移的 ActiveMQ 连接的 ConnectionFactory 匹配，因此在某些情况下，可能会捕获像消费者这样的远程关闭资源。

最终，尽管您的代码仍需要处理 JMS 资源之外的潜在故障情况并在需要时重试，例如发送者看到安全异常等。池位不会让您的所有麻烦都消失，在某些情况下他们只是介绍了您还没有想到的新产品。