项目场景:
测试RabbitMQ 消息确认机制中的ReturnCallback。已经按照要求做了如下配置:
- application.yml中配置:
spring:
rabbitmq:
host: 远程ip
port: 5672
virtual-host: /xzk
username: ******
password: ******
publisher-confirm-type: correlated
publisher-returns: true
template:
mandatory: true
问题描述:
使用Junit单元测试工具进行测试。
使用RabbitTemplate进行错误的消息发送后(具体是指发送到错误的routing key),没有成功执行ReturnCallback。但是使用调试模式执行的时候,却能正常执行ReturnCallback。
package com.kkb.hd;
import com.kkb.hd.config.RabbitMQConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.PropertySource;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;
/**
* @className: ProducerApplicationTest
* @description: TODO 类描述
* @author: HanDing
* @date: 2022/2/25
**/
@RunWith(SpringRunner.class)
@SpringBootTest
public class ProducerApplicationTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testReturn(){
rabbitTemplate.setMandatory(true); //此处和spring.rabbitmq.template.mandatory=true效果一样
rabbitTemplate.setReturnsCallback(returnedMessage -> {
System.out.println("return执行了...");
String exchange = returnedMessage.getExchange();
String routingKey = returnedMessage.getRoutingKey();
//String queue = returnedMessage.getMessage().getMessageProperties().getConsumerQueue();
System.out.println("消息从" + exchange + "到路由key为" + routingKey);
System.out.println("消息为:" + new String(returnedMessage.getMessage().getBody(), StandardCharsets.UTF_8));
});
//测试发送到一个不存在的routing key
rabbitTemplate.convertAndSend("spring-direct-exchange", "kkkk", "新增商品" );
}
}
原因分析:
ReturnCallback是在exchange->queue过程中发送失败才执行的。个人猜测exchange向queue发送消息与主线程是异步执行的,即主线程只要向exchange发送了消息就不管了,主线程继续后续的操作,而exchange在另一个线程中向queue发送消息。因此主线程提前结束,自然就执行不到exchange线程发送错误后的回调函数。
解决方案:
- 检查有没有配置
rabbitTemplate.setMandatory(true)
- 检查是否在application.yml中配置:
spring.rabbitmq.publisher-returns=true
(使用yaml的语法格式) - 以上确认无误后,在主线程最后加一段延时程序:
//测试发送到一个不存在的routing key
rabbitTemplate.convertAndSend("spring-direct-exchange", "kkkk", "新增商品" );
try {
TimeUnit.SECONDS.sleep(1L);
} catch (InterruptedException e) {
e.printStackTrace();
}