版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/M_Joes_17/article/details/81094715
1.需要将Template模版的publisherConfirms或publisherReturns的属性设置为true,此外ConnectionFactory要配置为CachingConnectionFactory的实现类
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost(host);
connectionFactory.setPort(port);
connectionFactory.setUsername(username);
connectionFactory.setPassword(password);
/**消息确认*/
connectionFactory.setPublisherConfirms(true);
/**消息回调*/
connectionFactory.setPublisherReturns(true);
return connectionFactory;
}
2.需设置mandatory=true,否则不回回调,消息就丢了
@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionFactory());
template.setMandatory(true);
template.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean b, String s) {
System.out.println("消息确认 ===== ack,是否确认 : " + b + ", 错误原因 : " + s);
if(correlationData != null){
System.out.println("correlationData : " + correlationData.toString());
}
}
});
template.setReturnCallback(new RabbitTemplate.ReturnCallback() {
@Override
public void returnedMessage(Message message, int i, String s, String s1, String s2) {
System.out.println("消息回调机制 ====== i :" + i + " , s(错误信息) : " + s + " , s1(交换机) : " + s1 + ", s2(队列) :" + s2);
if(message != null){
System.out.println("消息回调机制: message :" + message.toString());
}
}
});
return template;
}
3.消息确认的测试用例
@Component
public class RabbitMQSender {
private String queueName = "firstQueue";
private String routingkey = "first";
@Autowired
private RabbitAdmin admin;
@Autowired
private RabbitTemplate template;
public void send() throws IOException, TimeoutException {
System.out.println("开始发送消息....");
AMQP.BasicProperties persistentTextPlain = com.rabbitmq.client.MessageProperties.PERSISTENT_TEXT_PLAIN;
System.out.println("success:");
template.convertAndSend("direct", routingkey, "send message by exchange and routingkey..");
System.out.println("error:rountingkey");
template.convertAndSend("direct", routingkey + "1", "send message by exchange and routingkey..");
System.out.println("error:交换机");
template.convertAndSend("direct1", routingkey, "send message by exchange and routingkey..");
}
}
4.结论
- 如果消息没有到exchange,则confirm回调,ack=false
- 如果消息到达exchange,则confirm回调,ack=true
- exchange到queue成功,则不回调return
- exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)