我的 Spring 应用程序使用 ActiveMQ 队列。有两种可能的方法。这两种方法的 ActiveMQ 集成的初始部分是相同的:
@Bean
public ConnectionFactory connectionFactory() {
return new ActiveMQConnectionFactory();
}
@Bean
public Queue notificationQueue() {
return resolveAvcQueueByJNDIName("java:comp/env/jms/name.not.important.queue");
}
单线程方法:
@Bean
public IntegrationFlow orderNotify() {
return IntegrationFlows.from(Jms.inboundAdapter(connectionFactory()).destination(notificationQueue()),
c -> c.poller(Pollers.fixedDelay(QUEUE_POLLING_INTERVAL_MS)
.errorHandler(e -> logger.error("Can't handle incoming message", e))))
.handle(...).get();
}
但是我想使用多个工作线程来使用消息,因此我将代码从入站适配器重构为消息驱动的 channel 适配器:
@Bean
public IntegrationFlow orderNotify() {
return IntegrationFlows.from(Jms.messageDriverChannelAdapter(connectionFactory()).configureListenerContainer(c -> {
final DefaultMessageListenerContainer container = c.get();
container.setMaxConcurrentConsumers(notifyThreadPoolSize);
}).destination(notificationQueue()))
.handle(...).get();
}
问题在于,当应用程序重新部署到 Tomcat 或为第二种方法重新启动时,它不会停止 ActiveMQ 的使用者。它在启动期间创建新的消费者。但是所有新消息都被路由到旧的“死”消费者,因此它们位于“待处理消息”部分并且永远不会出队。
这里可能有什么问题?
最佳答案
我相信你必须完全停止 Tomcat。通常,在重新部署应用程序期间,Spring 容器应该停止并正确清除,但看起来这不是您的情况:Tomcat 重新部署 Hook 缺少一些内容。因此我建议完全停止它。
另一个选择是忘记外部 Tomcat,而只是迁移到 Spring Boot 并利用其启动嵌入式 servlet 容器的能力。这样在重建和重新启动应用程序后就不会出现任何泄漏。
关于java - ActiveMQ 队列的活跃消费者太多,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/52446229/