【RabbitMQ】消息的可靠性投递与签收

消息的可靠性投递——Porducer Confirm

确认模式

发生在从Producer发送到Exchange时，发送成功/失败都会自动调用ConfirmCallBack回调方法。

步骤：

（1）开启确认模式（在connectionFactory中设置publisher-confirms=“true”）

（2）编写ConfirmCallBack回调方法（rabbitTemplate模板上设置并编写回调方法）

回退模式

发生在从Exchange路由到Queue时，路由失败才会自动调用ReturnCallBack回调方法。

步骤：

（1）开启回退模式（在connectionFactory中设置publisher-returns=“true”）

（2）编写ReturnCallBack回调方法（rabbitTemplate模板上设置并编写回调方法）

（3）设置失败后的处理方式（false表示丢弃消息，true表示将消息返还给发送方）

代码

别忘了在connectionFactory中设置publisher-cofirms="true"或publisher-returns="true"

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:producer.xml")
public class ProducerTest {@Autowiredprivate RabbitTemplate rabbitTemplate;/***【确认模式】* 1.开启确认模式（在connectionFactory中设置publisher-confirms="true"）* 2.编写ConfirmCallBack回调方法（rabbitTemplate模板上设置并编写回调方法）*/@Testpublic void testConfirm() {// 设置confirm回调rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {public void confirm(CorrelationData correlationData, boolean ack, String cause) {System.out.println("confirm回调被执行了");if (ack) {System.out.println("接收成功！！！");} else {System.out.println("接收失败！！！");System.out.println("失败的原因是：" + cause);}}});// 发送消息rabbitTemplate.convertAndSend("test_exchange", "routing_key", "我是一个消息～");}/***【回退模式】* 1.开启回退模式（在connectionFactory中设置publisher-returns="true"）* 2.编写ReturnCallBack回调方法（rabbitTemplate模板上设置并编写回调方法）* 3.设置失败后的处理方式：*      - 设置为false：丢弃消息（默认）*      - 设置为true：返回消息给发送方*/@Testpublic void testReturn() {// 设置return回调rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {System.out.println("return回调被执行了");System.out.println("返回来的消息是：" + message);}});// 设置失败处理方式rabbitTemplate.setMandatory(true);// 发送消息rabbitTemplate.convertAndSend("test_exchange", "routing_key", "我是一个消息～");}}

消息的可靠性签收——Consumer Ack

ack即acknowledge确认属性，它可以在rabbit:listener-container标签中设置。

ack的值有3种：

（1）none：自动确认（自动签收）

（2）manual：手动确认（手动签收）

（3）auto：根据异常情况确认，比较麻烦

自动签收

消息一旦被Consumer接收到，则自动确认，并将该message从消息队列中移除。

这样做的缺点时显而易见的，当在业务处理时出错，你想要再次取出该meassge重新处理时，它已经不见了。

手动签收

消息被Consumer拿到后，不会立即从消息队列中移除，而是等待业务逻辑处理完毕并成功后，再由我们手动将其确认并移除。

代码

因为自动签收(channel.basicAck)和手动签收(channel.basicNack)都需要由管道对象(channel)调用
而MessageListener接口的抽象方法的参数中只有message而没有channel，所以这里要使用它的子接口ChannelAwareMessageListener

/*** 1.设置手动签收。在rabbit:listener-container中设置acknowledge="manual"* 2.编写一个实现ChannelAwareMessageListener接口的监听器* 3.如果消息成功处理，则调用channel的basicAck()方法手动签收* 4.如果消息成功失败，则调用channel的basicNack()方法拒绝签收，消息重回队列并再次重发*/@Component
public class AckListener implements ChannelAwareMessageListener {@Overridepublic void onMessage(Message message, Channel channel) throws Exception {// 转换并打印消息System.out.println(new String(message.getBody()));// 获取签收标签long deliveryTag = message.getMessageProperties().getDeliveryTag();try {System.out.println("处理业务逻辑...");channel.basicAck(deliveryTag, true);            // 手动签收} catch (Exception e) {channel.basicNack(deliveryTag ,true, true);     // 拒绝签收}}@Overridepublic void onMessage(Message message) {}
}