RabbitMQ问题系列:(一)_org.springframework.amqp.rabbit.support.listenerex

1 监听方法不能有返回值

• 警告信息：
  org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener threw exception
  Caused by: org.springframework.amqp.rabbit.listener.adapter.ReplyFailureException: Failed to send reply with payload ‘InvocationResult [returnValue=“stringfadfadfa”, returnType=class java.lang.String, bean=com.company.microservicedata.service.impl.MessageQueueReceiveServiceImpl@34abfdbf, method=public java.lang.String com.company.microservicedata.service.impl.MessageQueueReceiveServiceImpl.topicManyReceive(org.springframework.amqp.core.Message,com.rabbitmq.client.Channel) throws java.io.IOException]’
  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.
• 原因
  消费方法（监听消息并接收消息的方法）使用了返回值。
  源码：org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener#doHandleResult中调用的org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener#getReplyToAddress方法抛出的异常，其中responseAddress为null，触发AmqpException。

org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener#doHandleResult

org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener#getReplyToAddress

• 方案
  监听方法无返回值，void，样例如下：

@Override
@RabbitListener(queues="queue_n")
public void topicManyReceive(Message msg, Channel channel) throws IOException {
    try {
        logger.info("接收成功--Topic交换机queue_n: {}", msg);
        channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
    } catch(Exception e) {
        channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
        logger.info("接收失败: {}", e);
    }
}
1
2
3
4
5
6
7
8
9
10
11

2 不能多次ACK

• 错误信息
  Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 1, class-id=60, method-id=80)
  Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 1, class-id=60, method-id=80)
• 原因
  多次调用ACK，即配置了自动ACK之后，又手动调用了basicAck方法，两次确认消息
• 方案
  一次确认消息，即，配置了自动确认，禁止手动调用basicAck方法，配置手动确认，则需要主动调用确认方法basicAck方法
  （1）自动确认
  配置文件：

rabbitmq:
    host: 192.168.1.5
    port: 5672
    username: xindaqi
    password: 123456
    publisher-returns: true
    listener:
      direct:
        acknowledge-mode: auto
      simple: 
        acknowledge-mode: auto
        concurrency: 1 # 当前监听容器数
        max-concurrency: 1 # 最大监听数
        #retry:
        #  enabled: true # 支持重试
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

@Override
@RabbitListener(queues="queue_n")
public void topicManyReceive(Message msg, Channel channel) throws IOException {
    try {
        logger.info("接收成功--Topic交换机queue_n: {}", msg);
    } catch(Exception e) {
        logger.info("接收失败: {}", e);
    }
}
1
2
3
4
5
6
7
8
9

（2）手动确认
配置信息：

rabbitmq:
    host: 192.168.1.5
    port: 5672
    username: xindaqi
    password: 123456
    publisher-returns: true
    listener:
      direct:
        acknowledge-mode: manual
      simple: 
        acknowledge-mode: manual
        concurrency: 1 # 当前监听容器数
        max-concurrency: 1 # 最大监听数
        #retry:
        #  enabled: true # 支持重试
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

监听消息：

@Override
@RabbitListener(queues="queue_n")
public void topicManyReceive(Message msg, Channel channel) throws IOException {
    try {
        logger.info("接收成功--Topic交换机queue_n: {}", msg);
        channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
    } catch(Exception e) {
        channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
        logger.info("接收失败: {}", e);
    }
}
1
2
3
4
5
6
7
8
9
10
11

---

【参考文献】
[1]https://blog.csdn.net/qq_15071263/article/details/99976534
[2]https://blog.csdn.net/chenzt8812/article/details/100630565/
[3]https://blog.csdn.net/wxb880114/article/details/105854584