我正在使用SpringBoot启动连接到RabbitMQ队列的SpringAMQP应用程序。我希望能够从生产者发送一条消息,指定答复队列,以便消费者只需要发送而不必调查目的地(因此不必在消息本身中传递答复数据)。
这是我的配置(在生产者和消费者之间共享)
private static final String QUEUE_NAME = "testQueue";
private static final String ROUTING_KEY = QUEUE_NAME;
public static final String REPLY_QUEUE = "replyQueue";
private static final String USERNAME = "guest";
private static final String PASSWORD = "guest";
private static final String IP = "localhost";
private static final String VHOST = "/";
private static final int PORT = 5672;
@Bean
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionFactory());
amqpAdmin().declareQueue(new Queue(QUEUE_NAME));
amqpAdmin().declareQueue(new Queue(REPLY_QUEUE));
return template;
}
@Bean
public AmqpAdmin amqpAdmin() {
return new RabbitAdmin(connectionFactory());
}
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory(IP);
connectionFactory.setUsername(USERNAME);
connectionFactory.setPassword(PASSWORD);
connectionFactory.setVirtualHost(VHOST);
connectionFactory.setPort(PORT);
return connectionFactory;
}
我正在发送以下消息:
public Object sendAndReply(String queue, String content){
return template.convertSendAndReceive(queue, new Data(content), new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setReplyTo(ReplyTester.REPLY_QUEUE);
return message;
}
});
}
并等待以下答复:
public void replyToQueue(String queue){
template.receiveAndReply(queue, new ReceiveAndReplyCallback<Data, Data>() {
@Override
public Data handle(Data payload) {
System.out.println("Received: "+payload.toString());
return new Data("This is a reply for: "+payload.toString());
}
});
}
但是,在发送时,出现以下异常:
Exception in thread "main" org.springframework.amqp.UncategorizedAmqpException: java.lang.IllegalArgumentException: Send-and-receive methods can only be used if the Message does not already have a replyTo property.
at org.springframework.amqp.rabbit.support.RabbitExceptionTranslator.convertRabbitAccessException(RabbitExceptionTranslator.java:66)
at org.springframework.amqp.rabbit.connection.RabbitAccessor.convertRabbitAccessException(RabbitAccessor.java:112)
at org.springframework.amqp.rabbit.core.RabbitTemplate.doExecute(RabbitTemplate.java:841)
at org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:820)
at org.springframework.amqp.rabbit.core.RabbitTemplate.doSendAndReceiveWithTemporary(RabbitTemplate.java:705)
at org.springframework.amqp.rabbit.core.RabbitTemplate.doSendAndReceive(RabbitTemplate.java:697)
at org.springframework.amqp.rabbit.core.RabbitTemplate.convertSendAndReceive(RabbitTemplate.java:673)
at org.springframework.amqp.rabbit.core.RabbitTemplate.convertSendAndReceive(RabbitTemplate.java:663)
at prodsend.Prod.sendAndReply(ReplyTester.java:137)
at prodsend.ReplyTester.sendMessages(ReplyTester.java:49)
at prodsend.ReplyTester.main(ReplyTester.java:102)
Caused by: java.lang.IllegalArgumentException: Send-and-receive methods can only be used if the Message does not already have a replyTo property.
at org.springframework.util.Assert.isNull(Assert.java:89)
at org.springframework.amqp.rabbit.core.RabbitTemplate$6.doInRabbit(RabbitTemplate.java:711)
at org.springframework.amqp.rabbit.core.RabbitTemplate$6.doInRabbit(RabbitTemplate.java:705)
at org.springframework.amqp.rabbit.core.RabbitTemplate.doExecute(RabbitTemplate.java:835)
... 8 more
行ReplyTest.137指向上面
return
方法中的sendAndReply
行。编辑:
这是上面提到的Data类:)
class Data{
public String d;
public Data(String s){ d = s; }
public String toString() { return d; }
}
最佳答案
从documentation:
基本的RPC模式。使用特定的路由密钥将消息发送到默认交换机,并尝试接收响应。实现通常会将回复标头设置为互斥队列,并等待超时限制的一段时间。
因此,方法convertSendAndReceive
处理设置replyTo
标头并返回Messaage
-响应。这是一个同步模式-RPC。
如果要异步执行(似乎要这样做),请不要使用此方法。使用适当的convertAndSend
方法,并使用适当的MessagePostProcessor
添加您的replyTo
标头。
由于这是异步的,因此您需要注册一个单独的处理程序以接收答复。这需要在将消息发送给另一方之前完成。然后,在发送消息后的某个时候将调用此处理程序-未知时。阅读Spring AQMP Documentation的3.5.2异步使用方。
因此,异步处理流程:
发送方在replyTo
队列上注册处理程序
发件人发送replyTo
设置发送邮件
客户端调用receiveAndReply
,处理消息,然后将回复发送给replyTo
发送者回调方法已触发
同步处理流程为:
发件人使用sendAndReceive
发送消息并阻止
客户端调用receiveAndReply
,处理消息,然后将回复发送给replyTo
发件人收到答复,唤醒并处理它
因此,后一种情况要求发送者等待。当您使用receiveXXX
而不是注册异步处理程序时,如果客户端需要一些时间来解决receiveXXX
的问题,发送方可能会等待很长时间。
顺便说一句,如果要使用同步方法但使用特定的replyTo
,则可以随时调用setReplyQueue
。对于我提到的情况,还有一个setReplyTimeout
,其中客户端要么不打扰阅读邮件,要么忘记回复。