spring jmsListener to listen to multiple Queues

Question

In this post , Garry Russell explained how to programmatically create multiple KafkaListeners to listen from multiple topics.. [this setup is actually working successfully for me ] Kafka Spring: How to create Listeners dynamically or in a loop?

Now i want to have a similar setup working for JMSListeners as well - where I can have one class with one @JMSListener in it and i can programmatically create multiple instances of that JMSListener each injected with its own queueName.

I found this post Spring JMS start listening to jms queues on request

In the end of this post Gary made a similar comment,

If you wish to dynamically create lots of containers, then just create the containers programmatically, call afterPropertiesSet(), then start()

I used the setup that i had working from the first post above(related to KafkaListeners), my multiple instances of JMS listeners are starting up but not consuming any messages.

Basically i didn't understand where do i do this

then just create the containers programmatically, call afterPropertiesSet(), then start()

I am confused with the word - container, I know there's JMSListener and there's JmsListenerContainerFactory, what is a container in this context - i guess JMSListener?

i have confirmed there are messages in the queue. also when i don't create the listeners programmatically and just have one listener with hardcoded queue mentioned on it, it consumes the message fine.

Basically none of the listeners are consuming the messages when i am creating multiple JMS Listeners programmatically

    @SpringBootApplication
@EnableJms
public class MqProdConsumerApplication {
    private static Logger logger = LogManager.getLogger(MqProdConsumerApplication.class.getName());
    private static Consumers consumersStatic;

    @Autowired
    Consumers consumers;

    @PostConstruct
    public void init() {
        consumersStatic = this.consumers;
    }

    @Bean
    public Gson gson() {
        return new Gson();
    }

    public static void main(String[] args) {
        ConfigurableApplicationContext context = SpringApplication.run(MqProdConsumerApplication.class, args);
        List<QueueInformation> queueInformationList = consumersStatic.getQueueInformationList();
        Assert.notEmpty(queueInformationList, "queueInformationList cannot be empty");
        logger.debug("queueInformationList ************" + queueInformationList.toString());
        for (QueueInformation queueInformation : queueInformationList) {
            AnnotationConfigApplicationContext child = new AnnotationConfigApplicationContext();
            child.setParent(context);
            child.register(MQConfig.class);
            Properties props = new Properties();
            props.setProperty("mqQueueName", queueInformation.getMqQueueName());
            //
            PropertiesPropertySource pps = new PropertiesPropertySource("listenerProps", props);
            child.getEnvironment().getPropertySources().addLast(pps);
            child.refresh();
        }
    }
}

Here's the MQConfig that has the listenerContainerFactory

@Configuration
public class MQConfig {
    Logger logger = LoggerFactory.getLogger(this.getClass());

    @Value("${ibm.mq.user}")
    private String mqUserName;

    @Bean
    public MQListener listener() {
        return new MQListener();
    }

    @PostConstruct
    public void afterConstruct() {
        logger.debug("************* initialized MQ Config successfully for user =" + mqUserName);
    }

    @Bean
    public JmsListenerContainerFactory<?> myFactory(ConnectionFactory connectionFactory,
                                                    DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        configurer.configure(factory, connectionFactory);

        // Put the MQ username in the PCF environment.
        // Otherwise, the connection is identified by PCF's default user, "VCAP"
        System.setProperty("user.name", mqUserName);
        return factory;
    }
}

Then comes the MQListener which has the actual @JMSListener

    public class MQListener {
    Logger logger = LoggerFactory.getLogger(this.getClass());

    @Value("${mqQueueName}")
    private String mqQueueName;

    @PostConstruct
    public void afteConstruct() {
        logger.debug("************* initialized MQ Listener successfully, will read from =" + mqQueueName);

    }

    @JmsListener(destination = "${mqQueueName}", containerFactory = "myFactory")
    public void receiveMessage(String receivedMessage) throws JAXBException, ExecutionException, InterruptedException {
        logger.debug("***********************************************receivedMessage:" + receivedMessage);
    }
}

Here's my application.yml

    ibm.mq.queueManager: ABCTOD01
ibm.mq.channel: QMD00.SERVER
ibm.mq.connName: mqdv1.devfg.ABC.com
ibm.mq.user: pmd0app1
ibm.mq.password:
consumers:
  queueInformationList:
  -
    mqQueueName: QMD00.D.SRF.PERSON.LITE.PHONE.LOAD
  -
    mqQueueName: QMD00.D.SRF.PERSON.PHONE.LOAD

Robin Bajaj Robin Bajaj · Accepted Answer · 2019-04-05T21:46:28

ok i found another post where Gary has answered what i was looking for Adding Dynamic Number of Listeners(Spring JMS)

Essentially here's the working solution for me. Great job @GaryRussell - I am a fan now :)

@Configuration
@EnableJms
public class AppConfig implements JmsListenerConfigurer {

    @Override
    public void configureJmsListeners(JmsListenerEndpointRegistrar registrar) {
        List<QueueInformation> queueInformationList = consumersStatic.getQueueInformationList();
        int i = 0;
        for (QueueInformation queueInformation :
                queueInformationList) {
            SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint();
            endpoint.setId("myJmsEndpoint-" + i++);
            endpoint.setDestination(queueInformation.getMqQueueName());
            endpoint.setMessageListener(message -> {
                logger.debug("***********************************************receivedMessage:" + message);
            });
            registrar.registerEndpoint(endpoint);
            logger.debug("registered the endpoint for queue" + queueInformation.getMqQueueName());
  }
}

See also https://docs.spring.io/spring/docs/current/spring-framework-reference/integration.html#jms-annotated-programmatic-registration

spring jmsListener to listen to multiple Queues

1 Answers