Spring AMQP:由于缺少回复属性,从 POJO 侦听器发送回复失败

Posted

技术标签:

【中文标题】Spring AMQP:由于缺少回复属性,从 POJO 侦听器发送回复失败【英文标题】:Spring AMQP: Send reply from POJO listener failure because of missing reply-to property 【发布时间】:2016-03-18 10:00:38 【问题描述】:

我想编写一个简单的代码示例,它将消息发送到交换器,通过侦听器接收它并发送回消息。

这是我发送消息的方式:

private static void pingpong(Object messageListener) throws Exception 
    ConnectionFactory cf = new CachingConnectionFactory("localhost");

    // set up the queue, exchange, binding on the broker
    RabbitAdmin admin = new RabbitAdmin(cf);
    Queue queue = new Queue(QUEUE);
    admin.declareQueue(queue);
    TopicExchange exchange = new TopicExchange(EXCHANGE);
    admin.declareExchange(exchange);
    admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with("foo.*"));

    // set up the listener and container
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cf);
    container.setMessageListener(messageListener);
    container.setQueueNames(QUEUE);
    container.start();

    // send something and receive result.
    RabbitTemplate template = new RabbitTemplate(cf);
    Object result = template.convertSendAndReceive(EXCHANGE, "foo.bar", "ping");
    System.out.println("Received: " + result);

监听器的代码:

public static class PingPongReceiver 
    public String handleMessage(String in) 
        System.out.println("Received: " + in);
        return "pong";
    

我是这样称呼它的:

pingpong(new MessageListenerAdapter(new PingPongReceiver()));

但是发送回复失败,有以下例外:

org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:865)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:760)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:680)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$001(SimpleMessageListenerContainer.java:93)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$1.invokeListener(SimpleMessageListenerContainer.java:183)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1345)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:661)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1096)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:1080)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$800(SimpleMessageListenerContainer.java:93)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1190)
    at java.lang.Thread.run(Thread.java:745)
Caused by: org.springframework.amqp.rabbit.listener.adapter.ReplyFailureException: Failed to send reply with payload 'pong'
    at org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener.handleResult(AbstractAdaptableMessageListener.java:213)
    at org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter.onMessage(MessageListenerAdapter.java:296)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:757)
    ... 10 more
Caused by: org.springframework.amqp.AmqpException: Cannot determine ReplyTo message property value: Request message does not contain reply-to property, and no default response Exchange was set.
    at org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener.getReplyToAddress(AbstractAdaptableMessageListener.java:289)
    at org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener.handleResult(AbstractAdaptableMessageListener.java:209)
    ... 12 more

我在弄清楚如何使用匿名、非持久队列发送回复时遇到问题。我错过了什么?

【问题讨论】:

【参考方案1】:

你所描述的毫无意义; RabbitTemplate.convertAndReceive() 设置replyTo(Direct reply-to 如果代理支持,否则为临时队列)。

我刚刚复制并粘贴了您的代码,对我来说效果很好......

Received: ping
Received: pong

也许您的队列中有一条旧的、陈旧的、没有回复的消息?

您可以使用管理 UI 清除队列。

我还用

测试了你的代码
Queue queue = new AnonymousQueue();
...
container.setQueueNames(queue.getName());

而且效果也很好。

顺便说一下,在手动创建 Spring Bean 时,建议您调用afterPropertiesSet(如果 bean 是初始化 bean)。

container.afterPropertiesSet();

最后,你应该停止容器并销毁连接:

container.stop();
cf.destroy();

【讨论】:

以上是关于Spring AMQP:由于缺少回复属性,从 POJO 侦听器发送回复失败的主要内容,如果未能解决你的问题,请参考以下文章

Spring-AMQP 和直接回复

Spring AMQP RPC消费者并抛出异常

为啥在 spring-amqp 中导入 AsyncRabbitTemplate

Spring Amqp sendAndReceive操作获取null但没有发生超时

由于嵌套大小写而缺少关键字错误

由于缺少 WebApp 库,在 Spring-boot-starter 项目中构建失败