2019獨角獸企業(yè)重金招聘Python工程師標準>>>

通過Publisher Confirms and Returns機制，生產(chǎn)者可以判斷消息是否發(fā)送到了exchange及queue，而通過消費者確認機制，Rabbitmq可以決定是否重發(fā)消息給消費者，以保證消息被處理。

1.什么是Publisher Confirms and Returns？

Delivery processing acknowledgements from consumers to RabbitMQ are known as acknowledgements in AMQP 0-9-1 parlance; broker acknowledgements to publishers are a protocol extension called publisher confirms.?
地址：http://www.rabbitmq.com/confirms.html

根據(jù)RabbitMq官網(wǎng)定義，rabbitmq代理(broker)對發(fā)布者(publishers)的確認被稱作發(fā)布者確認(publisher confirms)，這種機制是Rabbitmq對標準Amqp協(xié)議的擴展。因此通過這種機制可以確認消息是否發(fā)送給了目標。

2.如何通過Spring amqp來使用Publisher Confirms and Returns機制？

Confirmed and returned messages are supported by setting the CachingConnectionFactory’s publisherConfirms and publisherReturns properties to ‘true’ respectively.When these options are set, Channel s created by the factory are wrapped in an PublisherCallbackChannel, which is used to facilitate the callbacks. When such a channel is obtained, the client can register a PublisherCallbackChannel.Listener with the Channel. The PublisherCallbackChannel implementation contains logic to route a confirm/return to the appropriate listener. These features are explained further in the following sections.?
http://docs.spring.io/spring-amqp/docs/1.6.3.RELEASE/reference/html/_reference.html#cf-pub-conf-ret

通過Spring amqp文檔可以看到，要使用這種機制需要將Template模版的設publisherConfirms 或publisherReturns 屬性設置為true，此外ConnectionFactory要配置為CachingConnectionFactory。

<bean id="connectionFactory"class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory"><property name="host" value="192.168.2.133" /><property name="port" value="5672" /><property name="username" value="sun" /><property name="password" value="123456" /><property name="publisherConfirms" value="true" /><property name="publisherReturns" value="true" /></bean>

2.1 ConfirmCallback的使用及觸發(fā)的一種場景

import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.stereotype.Service;/*** @author wangzhongqiu* Created on 2017/10/31.* @description:繼承RabbitTemplate.ConfirmCallback,消息確認監(jiān)聽器*/ @Service public class ConfirmCallBackListener implements RabbitTemplate.ConfirmCallback {private Logger log = LoggerFactory.getLogger(CommonProducer.class);@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {log.info("收到回調(diào)，成功發(fā)送到broker");} }

2.2 ReturnCallback的使用及觸發(fā)的一種場景

import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.stereotype.Service;/*** @author wangzhongqiu* Created on 2017/10/31.* @description:繼承RabbitTemplate.ReturnCallback,消息發(fā)送失敗返回監(jiān)聽器*/ @Service public class ReturnCallBackListener implements RabbitTemplate.ReturnCallback {private Logger log = LoggerFactory.getLogger(CommonProducer.class);@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {log.info("收到回調(diào)");log.info("return--message:" + new String(message.getBody()) + ",replyCode:" + replyCode + ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey);} }

使用場景：

如果消息沒有到exchange,則confirm回調(diào),ack=false

如果消息到達exchange,則confirm回調(diào),ack=true

exchange到queue成功,則不回調(diào)return

exchange到queue失敗,則回調(diào)return(需設置mandatory=true,否則不回回調(diào),消息就丟了)

轉載于:https://my.oschina.net/xiaominmin/blog/1811701

創(chuàng)作挑戰(zhàn)賽新人創(chuàng)作獎勵來咯，堅持創(chuàng)作打卡瓜分現(xiàn)金大獎

總結

以上是生活随笔為你收集整理的【RabbitMQ】6、rabbitmq生产者的消息确认的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。