超时后收到回复
Reply received after timeout
我想用最新的 Spring 版本为 Spring AMQP 注册两个队列侦听器:
@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;
}
听众:
@Component
public class TransactionElavonSaleListener {
public TransactionResponseFactory transactionElavonSaleProcess(TransactionRequestFactory ro) {
..... do some heavy network request
return parseRawSuccessResponse(response);
}
}
当我删除其中一个 SimpleMessageListenerContainer 时它是工作文件但是当我使用这两种方法时我得到这个异常:
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.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(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.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(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
你知道为什么我有这两种方法时会出现这个异常吗?
EDITL 定义的侦听器:
@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;
}
}
发送对象:
TransactionResponseFactory processingPeply = (TransactionResponseFactory) processingTransactionElavonAuthorizeTemplate.convertSendAndReceive(
ContextServer.EXCHANGE_PROCESSING, ContextServer.ROUTING_KEY_PROCESSING_TRANSACTION_ELAVON, tf);
System.out.println("!!!!! Received PROCESSING_TRANSACTION " + processingPeply.getTransaction_id());
关于您的配置的观察(但是与您的问题无关)...
- 您
AmqpAdmin
中的 declare...
代码不是必需的。
- 您应该永远不要在 bean 定义阶段与代理交互 - 现在太早了。
- 它们不是必需的,因为管理员会找到
Queue
和 Exchange
bean,并在首次打开连接时自动为您声明它们。
是的,消息转换器setter不用于此类侦听器;应该在 MessageListenerAdapter
上设置消息转换器。
但是,默认情况下它会得到一个 SimpleMessageConverter
,所以这应该不是问题;此转换器处理序列化对象以及纯文本。
现在,针对您的实际问题;添加第二个容器不应该对客户端有任何影响;每个模板都有自己的回复容器,默认情况下使用直接回复,因此它们之间不会发生串扰(如果您使用命名回复队列可能会发生这种情况,但这里不是这种情况)。
我建议您打开 DEBUG 日志记录以了解发生了什么;如果您需要帮助分析它们; post 日志(来自客户端和服务器端),我会看一下。
编辑
您的绑定不正确:
@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);
}
您将两个队列绑定到具有相同路由键的同一个交换器 - RabbitMQ 将使用该路由键向两个队列发送消息,因此两个侦听器都会响应。
convertSendAndReceive
的默认超时为 5 秒。要修改 AmqpTemplate
的 属性 replyTimeout
的超时设置为更大的值(毫秒)。
https://docs.spring.io/spring-amqp/reference/html/#reply-timeout
我想用最新的 Spring 版本为 Spring AMQP 注册两个队列侦听器:
@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;
}
听众:
@Component
public class TransactionElavonSaleListener {
public TransactionResponseFactory transactionElavonSaleProcess(TransactionRequestFactory ro) {
..... do some heavy network request
return parseRawSuccessResponse(response);
}
}
当我删除其中一个 SimpleMessageListenerContainer 时它是工作文件但是当我使用这两种方法时我得到这个异常:
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.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(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.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(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
你知道为什么我有这两种方法时会出现这个异常吗?
EDITL 定义的侦听器:
@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;
}
}
发送对象:
TransactionResponseFactory processingPeply = (TransactionResponseFactory) processingTransactionElavonAuthorizeTemplate.convertSendAndReceive(
ContextServer.EXCHANGE_PROCESSING, ContextServer.ROUTING_KEY_PROCESSING_TRANSACTION_ELAVON, tf);
System.out.println("!!!!! Received PROCESSING_TRANSACTION " + processingPeply.getTransaction_id());
关于您的配置的观察(但是与您的问题无关)...
- 您
AmqpAdmin
中的declare...
代码不是必需的。- 您应该永远不要在 bean 定义阶段与代理交互 - 现在太早了。
- 它们不是必需的,因为管理员会找到
Queue
和Exchange
bean,并在首次打开连接时自动为您声明它们。
是的,消息转换器setter不用于此类侦听器;应该在 MessageListenerAdapter
上设置消息转换器。
但是,默认情况下它会得到一个 SimpleMessageConverter
,所以这应该不是问题;此转换器处理序列化对象以及纯文本。
现在,针对您的实际问题;添加第二个容器不应该对客户端有任何影响;每个模板都有自己的回复容器,默认情况下使用直接回复,因此它们之间不会发生串扰(如果您使用命名回复队列可能会发生这种情况,但这里不是这种情况)。
我建议您打开 DEBUG 日志记录以了解发生了什么;如果您需要帮助分析它们; post 日志(来自客户端和服务器端),我会看一下。
编辑
您的绑定不正确:
@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);
}
您将两个队列绑定到具有相同路由键的同一个交换器 - RabbitMQ 将使用该路由键向两个队列发送消息,因此两个侦听器都会响应。
convertSendAndReceive
的默认超时为 5 秒。要修改 AmqpTemplate
的 属性 replyTimeout
的超时设置为更大的值(毫秒)。
https://docs.spring.io/spring-amqp/reference/html/#reply-timeout