1

I want to disconnect the DefaultMessageListenerContainer for a queue. I am using dmlc.stop(), dmlc.shutdown(). At the time of conneciton 5 consumer threads get connected to queue. when I try to disconnect, 4 of the consumers get disconnected, but 1 consumer remains connected. (See screenshot at the end of thread).

Environment 1. ActiveMQ with AMQP 2. SpringJMS with ApacheQpid

Problem After calling destroy and stop method, there's still one consumer connected to the queue.

Required Solution

I want to know, how to cleanly disconnect a MessageListenerContainer with zero consumer connections to the queue.

Configurations and Code

    @Bean
public DefaultMessageListenerContainer getMessageContainer(ConnectionFactory amqpConnectionFactory, QpidConsumer messageConsumer){
    DefaultMessageListenerContainer listenerContainer = new DefaultMessageListenerContainer();
    listenerContainer.setConcurrency("5-20");
    listenerContainer.setRecoveryInterval(jmsRecInterval);
    listenerContainer.setConnectionFactory(new CachingConnectionFactory(amqpConnectionFactory));
    listenerContainer.setMessageListener(messageConsumer);
    listenerContainer.setDestinationName(destinationName);
    return listenerContainer;
}

private void stopListenerIfRunning() {
        DefaultMessageListenerContainer dmlc = (DefaultMessageListenerContainer) ctx.getBean("messageContainer");
        if (null != dmlc) {
            if(!dmlc.isRunning()){return;}
            dmlc.stop(new Runnable() {
                @Override
                public void run() {
                    logger.debug("Closed Listener Container for Connection {}", sub.getQueueName());
                    if (sub.getSubscriptionStatus() == SubscriptionStatus.DELETED
                            || sub.getSubscriptionStatus() == SubscriptionStatus.SUSPENDED_DELETE) {
                        listenerHandles.remove(sub.getQueueName());
                    }
                }
            });
            dmlc.destroy();
            dmlc.shutdown();
        }

    }

}

enter image description here

learner
  • 906
  • 2
  • 10
  • 39

1 Answers1

2

listenerContainer.setConnectionFactory(newCachingConnectionFactory(amqpConnectionFactory));

You need to destroy the CachingConnectionFactory.

You generally don't need a caching factory with the listener container since the sessions are long-lived; you definitely should not if you have variable concurrency; from the javadocs...

 * <p><b>Note: Don't use Spring's {@link org.springframework.jms.connection.CachingConnectionFactory}
 * in combination with dynamic scaling.</b> Ideally, don't use it with a message
 * listener container at all, since it is generally preferable to let the
 * listener container itself handle appropriate caching within its lifecycle.
 * Also, stopping and restarting a listener container will only work with an
 * independent, locally cached Connection - not with an externally cached one.

if you want the connection cached, use a SingleConnectionFactory or call setCacheConsumers(false) on the CCF.

Gary Russell
  • 166,535
  • 14
  • 146
  • 179
  • I am creating one connection factory and using it in two or more listeners. Is it a bad approach? Shall I use one connection factory per message listener container? – learner Jan 17 '18 at 15:00
  • It's ok, but the CCF caches consumers by default. It's best to use the `SingleConnectionFactory` for listener containers. – Gary Russell Jan 17 '18 at 15:18
  • But in this case, you are creating a new CCF `listenerContainer.setConnectionFactory(new CachingConnectionFactory(amqpConnectionFactory));`. Which is wrong. – Gary Russell Jan 18 '18 at 03:22