SpringBoot+RabbitMQ ，Ack機制

一、先扔一張圖

說明:

本文涵蓋了關於RabbitMQ很多方面的知識點，如：

訊息傳送確認機制

消費確認機制

訊息的重新投遞

消費冪等性，等等

這些都是圍繞上面那張整體流程圖展開的，所以有必要先貼出來，見圖知意

二、實現思路

簡略介紹163郵箱授權碼的獲取

編寫傳送郵件工具類

編寫RabbitMQ配置檔案

生產者發起呼叫

消費者傳送郵件

定時任務定時拉取投遞失敗的訊息，重新投遞

各種異常情況的測試驗證

拓展：使用動態代理實現消費端冪等性驗證和訊息確認（ack）

三、專案介紹

springboot版本2。1。5。RELEASE，舊版本可能有些配置屬性不能使用，需要以程式碼形式進行配置

RabbitMQ版本3。7。15

MailUtil：傳送郵件工具類

RabbitConfig： rabbitmq相關配置

TestServiceImpl：生產者，傳送訊息

MailConsumer：消費者，消費訊息，傳送郵件

ResendMsg：定時任務，重新投遞傳送失敗的訊息

說明：上面是核心程式碼， MsgLogService mapper xml等均未貼出，完整程式碼可以參考GitHub上的原始碼，地址在文末。

四、程式碼實現

1.163郵箱授權碼的獲取, 如圖:

該授權碼就是配置檔案spring。mail。password需要的密碼

2.pom

<！——mq——> org。springframework。boot spring-boot-starter-amqp <！——mail——> org。springframework。boot spring-boot-starter-mail

3.rabbitmq、郵箱配置

# rabbitmqspring。rabbitmq。host=localhostspring。rabbitmq。port=5672spring。rabbitmq。username=guestspring。rabbitmq。password=guest# 開啟confirms回撥 P -> Exchangespring。rabbitmq。publisher-confirms=true# 開啟returnedMessage回撥 Exchange -> Queuespring。rabbitmq。publisher-returns=true# 設定手動確認（ack） Queue -> Cspring。rabbitmq。listener。simple。acknowledge-mode=manualspring。rabbitmq。listener。simple。prefetch=100# mailspring。mail。host=smtp。163。comspring。mail。username=18621142249@163。comspring。mail。password=123456wangzaispring。mail。from=18621142249@163。comspring。mail。properties。mail。smtp。auth=truespring。mail。properties。mail。smtp。starttls。enable=truespring。mail。properties。mail。smtp。starttls。required=true

說明： password即授權碼， username和from要一致

4.表結構

CREATE TABLE `msg_log` （ `msg_id` varchar（255） NOT NULL DEFAULT ‘’ COMMENT ‘訊息唯一標識’， `msg` text COMMENT ‘訊息體， json格式化’， `exchange` varchar（255） NOT NULL DEFAULT ‘’ COMMENT ‘交換機’， `routing_key` varchar（255） NOT NULL DEFAULT ‘’ COMMENT ‘路由鍵’， `status` int（11） NOT NULL DEFAULT ‘0’ COMMENT ‘狀態： 0投遞中 1投遞成功 2投遞失敗 3已消費’， `try_count` int（11） NOT NULL DEFAULT ‘0’ COMMENT ‘重試次數’， `next_try_time` datetime DEFAULT NULL COMMENT ‘下一次重試時間’， `create_time` datetime DEFAULT NULL COMMENT ‘建立時間’， `update_time` datetime DEFAULT NULL COMMENT ‘更新時間’， PRIMARY KEY （`msg_id`）， UNIQUE KEY `unq_msg_id` （`msg_id`） USING BTREE） ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT=‘訊息投遞日誌’；

說明： exchange routing_key欄位是在定時任務重新投遞訊息時需要用到的

5.MailUtil

@Component@Slf4jpublic class MailUtil { @Value（“${spring。mail。from}”） private String from； @Autowired private JavaMailSender mailSender； /** * 傳送簡單郵件 * * @param mail */ public boolean send（Mail mail） { String to = mail。getTo（）；// 目標郵箱 String title = mail。getTitle（）；// 郵件標題 String content = mail。getContent（）；// 郵件正文 SimpleMailMessage message = new SimpleMailMessage（）； message。setFrom（from）； message。setTo（to）； message。setSubject（title）； message。setText（content）； try { mailSender。send（message）； log。info（“郵件傳送成功”）； return true； } catch （MailException e） { log。error（“郵件傳送失敗， to： {}， title： {}”， to， title， e）； return false； } }}

6.RabbitConfig

@Configuration@Slf4jpublic class RabbitConfig { @Autowired private CachingConnectionFactory connectionFactory； @Autowired private MsgLogService msgLogService； @Bean public RabbitTemplate rabbitTemplate（） { RabbitTemplate rabbitTemplate = new RabbitTemplate（connectionFactory）； rabbitTemplate。setMessageConverter（converter（））； // 訊息是否成功傳送到Exchange rabbitTemplate。setConfirmCallback（（correlationData， ack， cause） -> { if （ack） { log。info（“訊息成功傳送到Exchange”）； String msgId = correlationData。getId（）； msgLogService。updateStatus（msgId， Constant。MsgLogStatus。DELIVER_SUCCESS）； } else { log。info（“訊息傳送到Exchange失敗， {}， cause： {}”， correlationData， cause）； } }）； // 觸發setReturnCallback回撥必須設定mandatory=true，否則Exchange沒有找到Queue就會丟棄掉訊息，而不會觸發回撥 rabbitTemplate。setMandatory（true）； // 訊息是否從Exchange路由到Queue，注意：這是一個失敗回撥，只有訊息從Exchange路由到Queue失敗才會回撥這個方法 rabbitTemplate。setReturnCallback（（message， replyCode， replyText， exchange， routingKey） -> { log。info（“訊息從Exchange路由到Queue失敗： exchange： {}， route： {}， replyCode： {}， replyText： {}， message： {}”， exchange， routingKey， replyCode， replyText， message）； }）； return rabbitTemplate； } @Bean public Jackson2JsonMessageConverter converter（） { return new Jackson2JsonMessageConverter（）； } // 傳送郵件 public static final String MAIL_QUEUE_NAME = “mail。queue”； public static final String MAIL_EXCHANGE_NAME = “mail。exchange”； public static final String MAIL_ROUTING_KEY_NAME = “mail。routing。key”； @Bean public Queue mailQueue（） { return new Queue（MAIL_QUEUE_NAME， true）； } @Bean public DirectExchange mailExchange（） { return new DirectExchange（MAIL_EXCHANGE_NAME， true， false）； } @Bean public Binding mailBinding（） { return BindingBuilder。bind（mailQueue（））。to（mailExchange（））。with（MAIL_ROUTING_KEY_NAME）； }}

7.TestServiceImpl生產訊息

@Servicepublic class TestServiceImpl implements TestService { @Autowired private MsgLogMapper msgLogMapper； @Autowired private RabbitTemplate rabbitTemplate； @Override public ServerResponse send（Mail mail） { String msgId = RandomUtil。UUID32（）； mail。setMsgId（msgId）； MsgLog msgLog = new MsgLog（msgId， mail， RabbitConfig。MAIL_EXCHANGE_NAME， RabbitConfig。MAIL_ROUTING_KEY_NAME）； msgLogMapper。insert（msgLog）；// 訊息入庫 CorrelationData correlationData = new CorrelationData（msgId）； rabbitTemplate。convertAndSend（RabbitConfig。MAIL_EXCHANGE_NAME， RabbitConfig。MAIL_ROUTING_KEY_NAME， MessageHelper。objToMsg（mail）， correlationData）；// 傳送訊息 return ServerResponse。success（ResponseCode。MAIL_SEND_SUCCESS。getMsg（））； }}

8.MailConsumer消費訊息, 傳送郵件

@Component@Slf4jpublic class MailConsumer { @Autowired private MsgLogService msgLogService； @Autowired private MailUtil mailUtil； @RabbitListener（queues = RabbitConfig。MAIL_QUEUE_NAME） public void consume（Message message， Channel channel） throws IOException { Mail mail = MessageHelper。msgToObj（message， Mail。class）； log。info（“收到訊息： {}”， mail。toString（））； String msgId = mail。getMsgId（）； MsgLog msgLog = msgLogService。selectByMsgId（msgId）； if （null == msgLog || msgLog。getStatus（）。equals（Constant。MsgLogStatus。CONSUMED_SUCCESS）） {// 消費冪等性 log。info（“重複消費， msgId： {}”， msgId）； return； } MessageProperties properties = message。getMessageProperties（）； long tag = properties。getDeliveryTag（）； boolean success = mailUtil。send（mail）； if （success） { msgLogService。updateStatus（msgId， Constant。MsgLogStatus。CONSUMED_SUCCESS）； channel。basicAck（tag， false）；// 消費確認 } else { channel。basicNack（tag， false， true）； } }}

說明：其實就完成了3件事： 1。保證消費冪等性， 2。傳送郵件， 3。更新訊息狀態，手動ack

9.ResendMsg定時任務重新投遞傳送失敗的訊息

@Component@Slf4jpublic class ResendMsg { @Autowired private MsgLogService msgLogService； @Autowired private RabbitTemplate rabbitTemplate； // 最大投遞次數 private static final int MAX_TRY_COUNT = 3； /** * 每30s拉取投遞失敗的訊息，重新投遞 */ @Scheduled（cron = “0/30 * * * * ？”） public void resend（） { log。info（“開始執行定時任務（重新投遞訊息）”）； List msgLogs = msgLogService。selectTimeoutMsg（）； msgLogs。forEach（msgLog -> { String msgId = msgLog。getMsgId（）； if （msgLog。getTryCount（） >= MAX_TRY_COUNT） { msgLogService。updateStatus（msgId， Constant。MsgLogStatus。DELIVER_FAIL）； log。info（“超過最大重試次數，訊息投遞失敗， msgId： {}”， msgId）； } else { msgLogService。updateTryCount（msgId， msgLog。getNextTryTime（））；// 投遞次數+1 CorrelationData correlationData = new CorrelationData（msgId）； rabbitTemplate。convertAndSend（msgLog。getExchange（）， msgLog。getRoutingKey（）， MessageHelper。objToMsg（msgLog。getMsg（））， correlationData）；// 重新投遞 log。info（“第 ” + （msgLog。getTryCount（） + 1） + “ 次重新投遞訊息”）； } }）； log。info（“定時任務執行結束（重新投遞訊息）”）； }}

說明：每一條訊息都和exchange routingKey繫結，所有訊息重投共用這一個定時任務即可

五、基本測試

OK，目前為止，程式碼準備就緒，現在進行正常流程的測試

1.傳送請求:

2.後臺日誌:

3.資料庫訊息記錄:

狀態為3，表明已消費，訊息重試次數為0，表明一次投遞就成功了

4.檢視郵箱

傳送成功

六、各種異常情況測試

步驟一羅列了很多關於RabbitMQ的知識點，很重要，很核心，而本文也涉及到了這些知識點的實現，接下來就透過異常測試進行驗證（這些驗證都是圍繞本文開頭扔的那張流程圖展開的，很重要，所以，再貼一遍）

1.驗證訊息傳送到Exchange失敗情況下的回撥, 對應上圖P -> X

如何驗證？可以隨便指定一個不存在的交換機名稱，請求介面，看是否會觸發回撥

傳送失敗，原因： reply-code=404， reply-text=NOT_FOUND - no exchange ‘mail。exchangeabcd’ in vhost ‘/’，該回調能夠保證訊息正確傳送到Exchange，測試完成

2.驗證訊息從Exchange路由到Queue失敗情況下的回撥, 對應上圖X -> Q

同理，修改一下路由鍵為不存在的即可，路由失敗，觸發回撥

傳送失敗，原因： route： mail。routing。keyabcd， replyCode： 312， replyText： NO_ROUTE

3.驗證在手動ack模式下, 消費端必須進行手動確認(ack), 否則訊息會一直儲存在佇列中, 直到被消費, 對應上圖Q -> C

將消費端程式碼channel。basicAck（tag， false）；// 消費確認註釋掉，檢視控制檯和rabbitmq管控臺

可以看到，雖然訊息確實被消費了，但是由於是手動確認模式，而最後又沒手動確認，所以，訊息仍被rabbitmq儲存，所以，手動ack能夠保證訊息一定被消費，但一定要記得basicAck

4.驗證消費端冪等性

接著上一步，去掉註釋，重啟伺服器，由於有一條未被ack的訊息，所以重啟後監聽到訊息，進行消費，但是由於消費前會判斷該訊息的狀態是否未被消費，發現status=3，即已消費，所以，直接return，這樣就保證了消費端的冪等性，即使由於網路等原因投遞成功而未觸發回撥，從而多次投遞，也不會重複消費進而發生業務異常

5.驗證消費端發生異常訊息也不會丟失

很顯然，消費端程式碼可能發生異常，如果不做處理，業務沒正確執行，訊息卻不見了，給我們感覺就是訊息丟失了，由於我們消費端程式碼做了異常捕獲，業務異常時，會觸發： channel。basicNack（tag， false， true）；，這樣會告訴rabbitmq該訊息消費失敗，需要重新入隊，可以重新投遞到其他正常的消費端進行消費，從而保證訊息不被丟失

測試： send方法直接返回false即可（這裡跟丟擲異常一個意思）

可以看到，由於channel。basicNack（tag， false， true），未被ack的訊息（unacked）會重新入隊並被消費，這樣就保證了訊息不會走丟

6.驗證定時任務的訊息重投

實際應用場景中，可能由於網路原因，或者訊息未被持久化MQ就宕機了，使得投遞確認的回撥方法ConfirmCallback沒有被執行，從而導致資料庫該訊息狀態一直是投遞中的狀態，此時就需要進行訊息重投，即使也許訊息已經被消費了

定時任務只是保證訊息100%投遞成功，而多次投遞的消費冪等性需要消費端自己保證

我們可以將回調和消費成功後更新訊息狀態的程式碼註釋掉，開啟定時任務，檢視是否重投

可以看到，訊息會重投3次，超過3次放棄，將訊息狀態置為投遞失敗狀態，出現這種非正常情況，就需要人工介入排查原因

七、拓展：使用動態代理實現消費端冪等性驗證和消費確認（ack）

不知道大家發現沒有，在MailConsumer中，真正的業務邏輯其實只是傳送郵件mailUtil。send（mail）而已，但我們又不得不在呼叫send方法之前校驗消費冪等性，傳送後，還要更新訊息狀態為“已消費”狀態，並手動ack，實際專案中，可能還有很多生產者-消費者的應用場景，如記錄日誌，傳送簡訊等等，都需要rabbitmq，如果每次都寫這些重複的公用程式碼，沒必要，也難以維護，所以，我們可以將公共程式碼抽離出來，讓核心業務邏輯只關心自己的實現，而不用做其他操作，其實就是AOP

為達到這個目的，有很多方法，可以用spring aop，可以用攔截器，可以用靜態代理，也可以用動態代理，在這裡，我用的是動態代理

目錄結構如下：