CODEWITHSUNDEEP
javaconcurrencyspring-integrationactivemq-classic
user641652
user641652

Reputation: 65

Too many active consumers for ActiveMQ queue

My Spring app consumes ActiveMQ queue. There are two approaches possible. Initial part of ActiveMQ integration is the same for the both approaches:

@Bean
public ConnectionFactory connectionFactory() {
    return new ActiveMQConnectionFactory();
}

@Bean
public Queue notificationQueue() {
    return resolveAvcQueueByJNDIName("java:comp/env/jms/name.not.important.queue");
}

Single thread approach:

@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();
}

But I want to consume messages using the several worker threads, so I refactored the code from Inbound adapter to Message driven channel adapter:

@Bean
public IntegrationFlow orderNotify() {
    return IntegrationFlows.from(Jms.messageDriverChannelAdapter(connectionFactory()).configureListenerContainer(c -> {
                final DefaultMessageListenerContainer container = c.get();
                container.setMaxConcurrentConsumers(notifyThreadPoolSize);
            }).destination(notificationQueue()))
                           .handle(...).get();
}

The problem is that the app doesn't stop ActiveMQ's consumer when it being redeployed into Tomcat or being restarted for the second approach. It creates new consumer during it's startup. But all new messages are being routed to the old "dead" consumer so they're sat in the "Pending messages" section and never being dequeued.

What can be the problem here?

Upvotes: 1

Views: 564

Answers (1)

Artem Bilan
Artem Bilan

Reputation: 121272

You have to stop Tomcat fully, I believe. Typically during redeploy for the application, the Spring container should be stopped and clear properly, but looks like that's not your case: there is something missed for the Tomcat redeploy hook. Therefore I suggest to stop it fully.

Another option is to forget external Tomcat and just migrate to Spring Boot with its ability to start embedded servlet container. This way there is not going to be any leaks after rebuilding and restarting application.

Upvotes: 2

Related Questions