聽叔一句勸，消息隊列的水太深，你把握不住！

很多人在做架構(gòu)設(shè)計時往往會“過度設(shè)計”,簡單問題復(fù)雜化，上來就引一堆中間件，我想大概原因主要有下面兩點：

為了秀（學(xué)）技術(shù)而架構(gòu)

我們常說技術(shù)是為業(yè)務(wù)服務(wù)的，不能為了技術(shù)而技術(shù)，為了秀技術(shù)引入一堆復(fù)雜架構(gòu)這是要不得的。

考慮問題不全面，或者說廣度不夠，不知道如何簡單化

舉個栗子，假設(shè)有一個高并發(fā)的用戶平臺需要處理注冊（寫）及登錄查詢（讀）功能，在數(shù)據(jù)庫層做了主從同步。為了解決主從同步延時問題引入了一個Redis，想實現(xiàn)寫主庫的時候同時寫Redis，然后讀的時候直接讀Redis，這就避免了主從延時同步問題。這就是典型的考慮問題不全面，這樣雖然能解決主從延時問題，但是又會導(dǎo)致雙寫事務(wù)的產(chǎn)生，那為什么不直接把主從同步的方式改成強同步復(fù)制呢，這樣不是直接保證了一致性嗎？

那你可能會說改成強同步復(fù)制不是會增加響應(yīng)時間進而影響系統(tǒng)吞吐量嗎，那咱還可以對用戶做個分庫，多做幾個主從同步出來不就可以了嗎？

誒誒誒，跑題了，今天咱不是說消息隊列嗎？

哦，言歸正傳。今天我們說說消息隊列的問題，希望看完本文大家在引入消息隊列的時候先想一想，是不是一定要引入？引入消息隊列后產(chǎn)生的問題能不能解決？

消息隊列的作用

在微服務(wù)開發(fā)中我們經(jīng)常會引入消息中間件實現(xiàn)業(yè)務(wù)解耦，執(zhí)行異步操作， 現(xiàn)在讓我們來看看使用消息中間件的好處和弊端。

首先需要肯定是使用消息組件有很多好處，其中最核心的三個是：解耦、異步、削峰。

解耦：客戶端只要講請求發(fā)送給特定的通道即可，不需要感知接收請求實例的情況。

異步：將消息寫入消息隊列，非必要的業(yè)務(wù)邏輯以異步的方式運行，加快響應(yīng)速度。

削峰：消息中間件在消息被消費之前一直緩存消息，消息處理端可以按照自己處理的并發(fā)量從消息隊列中慢慢處理消息，不會一瞬間壓垮業(yè)務(wù)。

當(dāng)然消息中間件并不是銀彈，引入消息機制后也會有如下一些弊端：

潛在的性能瓶頸：消息代理可能會存在性能瓶頸。幸運的是目前主流的消息中間件都支持高度的橫向擴展。

潛在的單點故障：消息代理的高可用性至關(guān)重要，否則系統(tǒng)整體的可靠性將受到影響，幸運的是大多數(shù)消息中間件都是高可用的。

額外的操作復(fù)雜性：消息系統(tǒng)是一個必須獨立安裝、配置和運維的系統(tǒng)組件，增加了運維的復(fù)雜度。

這些弊端我們借助消息中間件本身提供的擴展、高可用能力可以解決，但是要真正用好消息中間件我們還需要關(guān)注可能會遇到的一些設(shè)計難題。

消息隊列的設(shè)計難題

處理并發(fā)和順序消息

在生產(chǎn)環(huán)境中為了提高消息處理的能力以及應(yīng)用程序的吞吐量，一般會將消費者部署多個實例節(jié)點。那么帶來的挑戰(zhàn)就是如何確保每個消息只被處理一次，并且是按照他們的發(fā)送順序來處理的。

例如：假設(shè)有3個相同的接收方實例從同一個點對點通道讀取消息，發(fā)送方按順序發(fā)布了 Order Created、Order Updated 和 Order Cancelled 這3個事件消息。簡單的消息實現(xiàn)可能就會同事講每個消息給不同的接收方。若由于網(wǎng)絡(luò)問題導(dǎo)致延遲，消息可能沒有按照他們發(fā)出時的順序被處理，這將導(dǎo)致奇怪的行為，服務(wù)實例可能在另一個服務(wù)器處理 Order Created 消息之前處理 Order Cancelled消息。

Kafka 使用的解決方案是使用分片（分區(qū)）通道。整體解決方案分為三個部分：

一個主題通道由多個分片組成，每個分片的行為類似一個通道。

發(fā)送方在消息頭部指定分片鍵如orderId，Kafka使用分片鍵將消息分配給特定的分片。

將接收方的多個實例組合在一起，并將他們視為相同的邏輯接收方（消費者組）。kafka將每個分片分配給單個接收器，它在接收方啟動和關(guān)閉時重新分配分片。

如上圖所示，每個Order事件消息都將orderId作為其分片鍵。特定訂單的每個事件都發(fā)布到同一個分片。而且該分片中的消息始終由同一個接收方實例讀取，因此這樣就能夠保證按順序處理這些消息。

處理重復(fù)消息

引入消息架構(gòu)必須要解決的另一個挑戰(zhàn)是處理重復(fù)消息。在理想情況下，消息代理應(yīng)該只傳遞一次消息，但保證消息有且僅有一次的消息傳遞的成本通常很高。相反，很多消息組件承諾至少保證成功傳遞一次消息。

在正常情況下，消息組件只會傳遞一次消息。但是當(dāng)客戶端、網(wǎng)絡(luò)或消息組件故障可能導(dǎo)致消息被多次傳遞。假設(shè)客戶端在處理消息后發(fā)送確認消息前，他的數(shù)據(jù)庫崩潰了，這時消息組件將再次發(fā)送未確認的消息，在數(shù)據(jù)庫重新啟動時向該客戶端發(fā)送。

處理重復(fù)消息有以下兩種不同的方法：

編寫冪等消息處理程序

跟蹤消息并丟棄重復(fù)項

如果應(yīng)用程序處理消息的邏輯是滿足冪等的，那么重復(fù)消息就是無害的。程序的冪等性是指，即使這個應(yīng)用被相同輸入?yún)?shù)多次重復(fù)調(diào)用時，也不會產(chǎn)生額外的效果。例如：取消一個已經(jīng)取消的訂單，就是一個冪等性操作。同樣，創(chuàng)建一個已經(jīng)存在的訂單操作也必是這樣。滿足冪等的消息處理程序可以被放心的執(zhí)行多次，只要消息組件在傳遞消息時保持相同的消息順序。

但是不幸的是，應(yīng)用程序通常不是冪等的。或者你現(xiàn)在正在使用的消息組件在重新傳遞消息時不會保留排序。重復(fù)或無序消息可能會導(dǎo)致錯誤。在這種情況下，你需要編寫跟蹤消息并丟棄重復(fù)消息的消息處理程序。

考慮一個授權(quán)消費者信用卡的消息處理程序。它必須為每個訂單僅執(zhí)行一次信用卡授權(quán)操作。這段應(yīng)用程序每次調(diào)用時都會產(chǎn)生不同的效果。如果重復(fù)消息導(dǎo)致消息處理程序多次執(zhí)行該邏輯，則應(yīng)用程序的行為將不正確。執(zhí)行此類應(yīng)用程序邏輯的消息處理程序必須通過檢測和丟棄重復(fù)消息而讓它成為冪等的。

一個簡單的解決方案是消息接收方使用 message id 跟蹤他已處理的消息并丟棄任何重復(fù)項。例如，在數(shù)據(jù)庫表中存儲它消費的每條消息的 message id。

當(dāng)接收方處理消息時，它將消息的 message id 作為創(chuàng)建和變更業(yè)務(wù)實體的事務(wù)的一部分記錄在數(shù)據(jù)表里。如上圖所示，接收方將包含message id 的行插入 PROCESSED_MESSAGE表。如果消息是重復(fù)的，則INSERT將失敗，接收方可以選擇丟棄該消息。

另一個解決方案是消息處理程序在應(yīng)用程序表，而不是專門表中記錄 message id。當(dāng)時用具有受限事務(wù)模型的NoSQL數(shù)據(jù)庫時，此方法特別有用，因為 NoSQL數(shù)據(jù)庫通常不支持將針對兩個表的更新作為數(shù)據(jù)庫事務(wù)。

處理事務(wù)性消息

服務(wù)通常需要在更新數(shù)據(jù)庫的事務(wù)中發(fā)布消息，數(shù)據(jù)庫更新和消息發(fā)送都必須在事務(wù)中進行，否則服務(wù)可能會更新數(shù)據(jù)庫然后在發(fā)送消息之前崩潰。

如果服務(wù)不以原子方式執(zhí)行者兩個操作，則類似的故障可能使系統(tǒng)處于不一致狀態(tài)。

接下來我們看一下常用的保證事務(wù)消息的兩種解決方案，最后再看看現(xiàn)代消息組件RocketMQ的事務(wù)性消息解決方案。

如果你的應(yīng)用程序正在使用關(guān)系型數(shù)據(jù)庫，要保證數(shù)據(jù)的更新和消息發(fā)送之間的事務(wù)可以直接使用事務(wù)性發(fā)件箱模式，Transactional Outbox。

此模式使用數(shù)據(jù)庫表作為臨時消息隊列。如上圖所示，發(fā)送消息的服務(wù)有個OUTBOX數(shù)據(jù)表，在進行INSERT、UPDATE、DELETE 業(yè)務(wù)操作時也會給OUTBOX數(shù)據(jù)表INSERT一條消息記錄，這樣可以保證原子性，因為這是基于本地的ACID事務(wù)。

OUTBOX表充當(dāng)臨時消息隊列，然后我們在引入一個消息中繼（MessageRelay）的服務(wù)，由他從OUTBOX表中讀取數(shù)據(jù)并發(fā)布消息到消息組件。

消息中繼的實現(xiàn)可以很簡單，只需要通過定時任務(wù)定期從OUTBOX表中拉取最新未發(fā)布的數(shù)據(jù)，獲取到數(shù)據(jù)后將數(shù)據(jù)發(fā)送給消息組件，最后將完成發(fā)送的消息從OUTBOX表中刪除即可。

另外一種保證事務(wù)性消息的方式是基于數(shù)據(jù)庫的事務(wù)日志，也就是所謂的數(shù)據(jù)變更捕獲，Change Data Capture，簡稱CDC。

一般數(shù)據(jù)庫在數(shù)據(jù)發(fā)生變更的時候都會記錄事務(wù)日志（Transaction Log），比如MySQL的binlog。事務(wù)日志可以簡單的理解成數(shù)據(jù)庫本地的一個文件隊列，它主要記錄按時間順序發(fā)生的數(shù)據(jù)庫表變更記錄。

這里我們利用alibaba開源的組件canal結(jié)合MySQL來說明下這種模式的工作原理。

更多操作說明可以參考官方文檔：https://github.com/alibaba/canal

canal工作原理

canal 模擬 MySQL slave 的交互協(xié)議，把自己偽裝成一個MySQL的 slave節(jié)點 ，向 MySQL master 發(fā)送dump 協(xié)議；

MySQL master 收到 dump 請求，開始推送 binary log 給 slave (即 canal )；

canal 解析 binary log 對象(原始為 byte 流)，然后可以將解析后的數(shù)據(jù)直接發(fā)送給消息組件。

Apache RocketMQ在4.3.0版中已經(jīng)支持分布式事務(wù)消息，RocketMQ采用了2PC的思想來實現(xiàn)了提交事務(wù)消息，同時增加一個補償邏輯來處理二階段超時或者失敗的消息，如下圖所示。

RocketMQ實現(xiàn)事務(wù)消息主要分為兩個階段：正常事務(wù)的發(fā)送及提交、事務(wù)信息的補償流程。

整體流程為：

正常事務(wù)發(fā)送與提交階段

1、生產(chǎn)者發(fā)送一個半消息給MQServer（半消息是指消費者暫時不能消費的消息）

2、服務(wù)端響應(yīng)消息寫入結(jié)果，半消息發(fā)送成功

3、開始執(zhí)行本地事務(wù)

4、根據(jù)本地事務(wù)的執(zhí)行狀態(tài)執(zhí)行Commit或者Rollback操作

事務(wù)信息的補償流程

1、如果MQServer長時間沒收到本地事務(wù)的執(zhí)行狀態(tài)會向生產(chǎn)者發(fā)起一個確認回查的操作請求

2、生產(chǎn)者收到確認回查請求后，檢查本地事務(wù)的執(zhí)行狀態(tài)

3、根據(jù)檢查后的結(jié)果執(zhí)行Commit或者Rollback操作

補償階段主要是用于解決生產(chǎn)者在發(fā)送Commit或者Rollback操作時發(fā)生超時或失敗的情況。

在生產(chǎn)者使用RocketMQ發(fā)送事務(wù)消息的時候我們也會借鑒第一種方案即自建一張事務(wù)日志表，然后在執(zhí)行本地事務(wù)的時候同時生成一條事務(wù)日志記錄，讓本地事務(wù)與日志事務(wù)在同一個方法中，同時添加 @Transactional 注解，保證兩個操作事務(wù)是一個原子操作。這樣如果事務(wù)日志表中有這個本地事務(wù)的信息，那就代表本地事務(wù)執(zhí)行成功，需要Commit，相反如果沒有對應(yīng)的事務(wù)日志，則表示沒執(zhí)行成功，需要Rollback。

孩砸，看完這篇文章，消息隊列你能把握住了嗎？

Kafka MySQL 數(shù)據(jù)庫

版權(quán)聲明：本文內(nèi)容由網(wǎng)絡(luò)用戶投稿，版權(quán)歸原作者所有，本站不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實的內(nèi)容，請聯(lián)系我們jiasou666@gmail.com 處理，核實后本網(wǎng)站將在24小時內(nèi)刪除侵權(quán)內(nèi)容。

版權(quán)聲明：本文內(nèi)容由網(wǎng)絡(luò)用戶投稿，版權(quán)歸原作者所有，本站不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實的內(nèi)容，請聯(lián)系我們jiasou666@gmail.com 處理，核實后本網(wǎng)站將在24小時內(nèi)刪除侵權(quán)內(nèi)容。