I want to register two queues listeners for Spring AMQP with the latest Spring version:
@Bean
public SimpleMessageListenerContainer processingTransactionSaleContainer(ConnectionFactory cf, TransactionElavonSaleListener listener) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cf);
container.setQueueNames(QUEUE_PROCESSING_SALE);
container.setMessageListener(new MessageListenerAdapter(listener, "transactionSaleProcess"));
container.setMessageConverter(new SerializerMessageConverter());
return container;
}
@Bean
public SimpleMessageListenerContainer processingTransactionAuthorizeContainer(ConnectionFactory cf, TransactionAuthorizeListener listener) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cf);
container.setQueueNames(QUEUE_PROCESSING_AUTHORIZE);
container.setMessageListener(new MessageListenerAdapter(listener, "transactionAuthorizeProcess"));
container.setMessageConverter(new SerializerMessageConverter());
return container;
}
Listener:
@Component
public class TransactionElavonSaleListener {
public TransactionResponseFactory transactionElavonSaleProcess(TransactionRequestFactory ro) {
..... do some heavy network request
return parseRawSuccessResponse(response);
}
}
When I remove one of the SimpleMessageListenerContainer it's working file but when I use the both methods I get this exception:
00:40:14,469 INFO [stdout] (pool-9-thread-5) 00:40:14.468 [pool-9-thread-5] WARN o.s.amqp.rabbit.core.RabbitTemplate - Reply received after timeout for 1
00:40:14,472 INFO [stdout] (pool-9-thread-5) 00:40:14.472 [pool-9-thread-5] WARN o.s.a.r.l.ConditionalRejectingErrorHandler - Execution of Rabbit message listener failed.
00:40:14,473 INFO [stdout] (pool-9-thread-5) org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
00:40:14,473 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:1613)
00:40:14,473 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1517)
00:40:14,473 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1440)
00:40:14,473 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1428)
00:40:14,473 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1423)
00:40:14,473 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1372)
00:40:14,473 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer$SimpleConsumer.callExecuteListener(DirectMessageListenerContainer.java:995)
00:40:14,473 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer$SimpleConsumer.handleDelivery(DirectMessageListenerContainer.java:955)
00:40:14,474 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//com.rabbitmq.client.impl.ConsumerDispatcher$5.run(ConsumerDispatcher.java:149)
00:40:14,474 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//com.rabbitmq.client.impl.ConsumerWorkService$WorkPoolRunnable.run(ConsumerWorkService.java:104)
00:40:14,474 INFO [stdout] (pool-9-thread-5) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1135)
00:40:14,474 INFO [stdout] (pool-9-thread-5) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
00:40:14,478 INFO [stdout] (pool-9-thread-5) at java.base/java.lang.Thread.run(Thread.java:844)
00:40:14,481 INFO [stdout] (pool-9-thread-5) Caused by: org.springframework.amqp.AmqpRejectAndDontRequeueException: Reply received after timeout
00:40:14,494 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.core.RabbitTemplate.onMessage(RabbitTemplate.java:2446)
00:40:14,494 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.lambda$setMessageListener$1(DirectReplyToMessageListenerContainer.java:115)
00:40:14,494 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1514)
00:40:14,494 INFO [stdout] (pool-9-thread-5) ... 11 common frames omitted
00:40:14,495 INFO [stdout] (pool-9-thread-5) 00:40:14.495 [pool-9-thread-5] ERROR o.s.a.r.l.DirectReplyToMessageListenerContainer - Failed to invoke listener
00:40:14,495 INFO [stdout] (pool-9-thread-5) org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
00:40:14,495 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:1613)
00:40:14,495 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1517)
00:40:14,496 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1440)
00:40:14,496 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1428)
00:40:14,496 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1423)
00:40:14,496 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1372)
00:40:14,496 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer$SimpleConsumer.callExecuteListener(DirectMessageListenerContainer.java:995)
00:40:14,496 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer$SimpleConsumer.handleDelivery(DirectMessageListenerContainer.java:955)
00:40:14,496 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//com.rabbitmq.client.impl.ConsumerDispatcher$5.run(ConsumerDispatcher.java:149)
00:40:14,496 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//com.rabbitmq.client.impl.ConsumerWorkService$WorkPoolRunnable.run(ConsumerWorkService.java:104)
00:40:14,496 INFO [stdout] (pool-9-thread-5) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1135)
00:40:14,496 INFO [stdout] (pool-9-thread-5) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
00:40:14,497 INFO [stdout] (pool-9-thread-5) at java.base/java.lang.Thread.run(Thread.java:844)
00:40:14,497 INFO [stdout] (pool-9-thread-5) Caused by: org.springframework.amqp.AmqpRejectAndDontRequeueException: Reply received after timeout
00:40:14,497 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.core.RabbitTemplate.onMessage(RabbitTemplate.java:2446)
00:40:14,497 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.lambda$setMessageListener$1(DirectReplyToMessageListenerContainer.java:115)
00:40:14,497 INFO [stdout] (pool-9-thread-5) at deployment.rest_api.war//org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1514)
00:40:14,497 INFO [stdout] (pool-9-thread-5) ... 11 common frames omitted
Do you know why I get this exception when I have these two methods?
EDITL Defined listeners:
@Bean
public SimpleMessageListenerContainer processingTransactionElavonSaleContainer(ConnectionFactory cf, TransactionElavonSaleListener listener) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cf);
container.setQueueNames(QUEUE_PROCESSING_ELAVON_SALE);
container.setMessageListener(new MessageListenerAdapter(listener, "transactionElavonSaleProcess"));
return container;
}
@Bean
public SimpleMessageListenerContainer processingTransactionElavonAuthorizeContainer(ConnectionFactory cf, TransactionElavonAuthorizeListener listener) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cf);
container.setQueueNames(QUEUE_PROCESSING_ELAVON_AUTHORIZE);
container.setMessageListener(new MessageListenerAdapter(listener, "transactionElavonAuthorizeProcess"));
return container;
}
...
@Component
public class TransactionElavonSaleListener {
public TransactionResponseFactory transactionElavonSaleProcess(TransactionRequestFactory ro) {
return new TransactionResponseFactory();
}
}
@Component
public class TransactionElavonAuthorizeListener {
public TransactionResponseFactory transactionElavonAuthorizeProcess(TransactionRequestFactory tf) {
TransactionResponseFactory obj = new TransactionResponseFactory();
return obj;
}
}
Send object:
TransactionResponseFactory processingPeply = (TransactionResponseFactory) processingTransactionElavonAuthorizeTemplate.convertSendAndReceive(
ContextServer.EXCHANGE_PROCESSING, ContextServer.ROUTING_KEY_PROCESSING_TRANSACTION_ELAVON, tf);
System.out.println("!!!!! Received PROCESSING_TRANSACTION " + processingPeply.getTransaction_id());
Observation about your configuration (not relevant to your issue, however)...
declare... code in your AmqpAdmin are not required.
Queue and Exchange beans and automatically declare them for you when the connection is first opened.Yes, that message converter setter is not used for this kind of listener; the message converter should be set on the MessageListenerAdapter instead.
However, it gets a SimpleMessageConverter by default so that should not be the problem; this converter handles serialized objects as well as plain text.
Now, to your actual issue; adding a second container should not have any affect on the client side; each template gets its own reply container and direct reply-to is used by default so there will be no cross talk between them (which can happen if you use a named reply queue, but that's not the case here).
I suggest you turn on DEBUG logging to figure out what's going on; if you need help analyzing them; post the logs (from both client and server side) and I'll take a look.
EDIT
Your bindings are incorrect:
@Bean
public Binding bindingQueueProcessingElavonSale() {
return BindingBuilder.bind(new Queue(QUEUE_PROCESSING_ELAVON_SALE, true))
.to(new TopicExchange(EXCHANGE_PROCESSING)).with(ROUTING_KEY_PROCESSING_TRANSACTION_ELAVON);
}
@Bean
public Binding bindingQueueProcessingElavonAuthorize() {
return BindingBuilder.bind(new Queue(QUEUE_PROCESSING_ELAVON_AUTHORIZE, true))
.to(new TopicExchange(EXCHANGE_PROCESSING)).with(ROUTING_KEY_PROCESSING_TRANSACTION_ELAVON);
}
You are binding both queues to the same exchange with the same routing key - RabbitMQ will send messages with that routing key to both queues, so both listeners will respond.