一、前情提示

上一篇文章《??百億級(jí)流量的系統(tǒng)架構(gòu)該怎么設(shè)計(jì)，今天就來教會(huì)你！??》分析了一下如何利用消息中間件對(duì)系統(tǒng)進(jìn)行解耦處理。

同時(shí)，我們也提到了使用消息中間件還有利于一份數(shù)據(jù)被多個(gè)系統(tǒng)同時(shí)訂閱，供多個(gè)系統(tǒng)來使用于不同的目的。

目前的一個(gè)架構(gòu)如下圖所示。

在這個(gè)圖里，我們可以清晰的看到，實(shí)時(shí)計(jì)算平臺(tái)發(fā)布的一份數(shù)據(jù)到消息中間件里，接著，會(huì)進(jìn)行如下步驟：

0. 數(shù)據(jù)查詢平臺(tái)，會(huì)訂閱這份數(shù)據(jù)，并落入自己本地的數(shù)據(jù)庫(kù)集群和緩存集群里，接著對(duì)外提供數(shù)據(jù)查詢的服務(wù)
1. 數(shù)據(jù)質(zhì)量監(jiān)控系統(tǒng)，會(huì)對(duì)計(jì)算結(jié)果按照一定的業(yè)務(wù)規(guī)則進(jìn)行監(jiān)控，如果發(fā)現(xiàn)有數(shù)據(jù)計(jì)算錯(cuò)誤，則會(huì)立馬進(jìn)行報(bào)警
2. 數(shù)據(jù)鏈路追蹤系統(tǒng)，會(huì)采集計(jì)算結(jié)果作為一個(gè)鏈路節(jié)點(diǎn)，同時(shí)對(duì)一條數(shù)據(jù)的整個(gè)完整計(jì)算鏈路都進(jìn)行采集并組裝出來一系列的數(shù)據(jù)計(jì)算鏈路落地存儲(chǔ)，最后如果某個(gè)數(shù)據(jù)計(jì)算錯(cuò)誤了，就可以立馬通過計(jì)算鏈路進(jìn)行回溯排查問題

因此上述場(chǎng)景中，使用消息中間件一來可以解耦，二來還可以實(shí)現(xiàn)消息“Pub/Sub”模型，實(shí)現(xiàn)消息的發(fā)布與訂閱。

這篇文章，咱們就來看看，假如說基于RabbitMQ作為消息中間件，如何實(shí)現(xiàn)一份數(shù)據(jù)被多個(gè)系統(tǒng)同時(shí)訂閱的“Pub/Sub”模型。

二、基于消息中間件的隊(duì)列消費(fèi)模型

上面那個(gè)圖，其實(shí)就是采用的RabbitMQ最基本的隊(duì)列消費(fèi)模型的支持。

也就是說，你可以理解為RabbitMQ內(nèi)部有一個(gè)隊(duì)列，生產(chǎn)者不斷的發(fā)送數(shù)據(jù)到隊(duì)列里，消息按照先后順序進(jìn)入隊(duì)列中排隊(duì)。

接著，假設(shè)隊(duì)列里有4條數(shù)據(jù)，然后我們有2個(gè)消費(fèi)者一起消費(fèi)這個(gè)隊(duì)列的數(shù)據(jù)。

此時(shí)每個(gè)消費(fèi)者會(huì)均勻的被分配到2條數(shù)據(jù)，也就是說4條數(shù)據(jù)會(huì)均勻的分配給各個(gè)消費(fèi)者，每個(gè)消費(fèi)者只不過是處理一部分?jǐn)?shù)據(jù)罷了，這個(gè)就是典型的隊(duì)列消費(fèi)模型。

之前這幾篇文章，基本給出了上述那個(gè)最基本的隊(duì)列消費(fèi)模型的RabbitMQ代碼實(shí)現(xiàn)，以及如何保證消費(fèi)者宕機(jī)時(shí)數(shù)據(jù)不丟失，如何讓RabbitMQ集群對(duì)queue和message都進(jìn)行持久化。基本上整體代碼實(shí)現(xiàn)都比較完整，大家可以參考一下。

三、基于消息中間件的“Pub/Sub”模型

但是消息中間件還可以實(shí)現(xiàn)一種“Pub/Sub”模型，也就是“發(fā)布/訂閱”模型，Pub就是Publish，Sub就是Subscribe。

這種模型是可以支持多個(gè)系統(tǒng)同時(shí)消費(fèi)一份數(shù)據(jù)的。也就是說，你發(fā)布出去的每條數(shù)據(jù)，都會(huì)廣播給每個(gè)系統(tǒng)。

給大家來一張圖，一起來感受一下。

如上圖所示。也就是說，我們想要實(shí)現(xiàn)的上圖的效果，實(shí)時(shí)計(jì)算平臺(tái)發(fā)布一系列的數(shù)據(jù)到消息中間件里。

然后數(shù)據(jù)查詢平臺(tái)、數(shù)據(jù)質(zhì)量監(jiān)控系統(tǒng)、數(shù)據(jù)鏈路追蹤系統(tǒng)，都會(huì)訂閱數(shù)據(jù)，都會(huì)消費(fèi)到同一份完整的數(shù)據(jù)，每個(gè)系統(tǒng)都可以根據(jù)自己的需要使用數(shù)據(jù)。

這，就是所謂的“Pub/Sub”模型，一個(gè)系統(tǒng)發(fā)布一份數(shù)據(jù)出去，多個(gè)系統(tǒng)訂閱和消費(fèi)到一模一樣的一份數(shù)據(jù)。

那如果要實(shí)現(xiàn)上述的效果，基于RabbitMQ應(yīng)該怎么來處理呢？

四、RabbitMQ中的exchange到底是個(gè)什么東西？

實(shí)際上來說，在RabbitMQ里面是不允許生產(chǎn)者直接投遞消息到某個(gè)queue（隊(duì)列）里的，而是只能讓生產(chǎn)者投遞消息給RabbitMQ內(nèi)部的一個(gè)特殊組件，叫做“exchange”。

關(guān)于這個(gè)exchange，大概你可以把這個(gè)組件理解為一種消息路由的組件。

也就是說，實(shí)時(shí)計(jì)算平臺(tái)發(fā)送出去的message到RabbitMQ中都是由一個(gè)exchange來接收的。

然后這個(gè)exchange會(huì)根據(jù)一定的規(guī)則決定要將這個(gè)message路由轉(zhuǎn)發(fā)到哪個(gè)queue里去，這個(gè)實(shí)際上就是RabbitMQ中的一個(gè)核心的消息模型。

大家看下面的圖，一起來理解一下。

五、默認(rèn)的exchange

在之前的文章里，我們投遞消息到RabbitMQ的時(shí)候，也沒有用什么exchange，但是為什么就還是把消息投遞到了queue里去呢？

那是因?yàn)槲覀冇昧四J(rèn)的exchange，他會(huì)直接把消息路由到你指定的那個(gè)queue里去，所以如果簡(jiǎn)單用隊(duì)列消費(fèi)模型，不就省去了exchange的概念了嗎。

上面這段就是之前我們給大家展示的，讓消息持久化的一種投遞消息的方式。

大家注意里面的第一個(gè)參數(shù)，是一個(gè)空的字符串，這個(gè)空字符串的意思，就是說投遞消息到默認(rèn)的exchange里去，然后他就會(huì)路由消息到我們指定的queue里去。

六、將消息投遞到fanout exchange

在RabbitMQ里，exchange這種組件有很多種類型，比如說：direct、topic、headers以及fanout。這里咱們就來看看最后一種，fanout這種類型的exchange組件。

這種exchange組件其實(shí)非常的簡(jiǎn)單，你可以創(chuàng)建一個(gè)fanout類型的exchange，然后給這個(gè)exchange綁定多個(gè)queue。

接著只要你投遞一條消息到這個(gè)exchange，他就會(huì)把消息路由給他綁定的所有queue。

使用下面的代碼就可以創(chuàng)建一個(gè)exchange，比如說在實(shí)時(shí)計(jì)算平臺(tái)（生產(chǎn)者）的代碼里，可以加入下面的一段，創(chuàng)建一個(gè)fanout類型的exchange。

第一個(gè)參數(shù)我們叫做“rt_compute_data”，這個(gè)就是exchange的名字，rt就是“RealTime”的縮寫，意思就是實(shí)時(shí)計(jì)算系統(tǒng)的計(jì)算結(jié)果數(shù)據(jù)。

第二個(gè)參數(shù)就是定義了這個(gè)exchange的類型是“fanout”。

channel.exchangeDeclare(
"rt_compute_data",
"fanout");

接著我們就采用下面的代碼來投遞消息到我們創(chuàng)建好的exchange組件里去：

大家會(huì)注意到，此時(shí)消息就是投遞到指定的exchange里去了，但是路由到哪個(gè)queue里去呢？此時(shí)我們暫時(shí)還沒確定，要讓消費(fèi)者自己來把自己的queue綁定到這個(gè)exchange上去才可以。

七、綁定自己的隊(duì)列到exchange上去消費(fèi)

我們對(duì)消費(fèi)者的代碼也進(jìn)行修改，之前我們?cè)谶@里關(guān)閉了autoAck機(jī)制，然后每次都是自己手動(dòng)ack。

上面的代碼里，每個(gè)消費(fèi)者系統(tǒng)，都會(huì)有一些不一樣，就是每個(gè)消費(fèi)者都需要定義自己的隊(duì)列，然后綁定到exchange上去。比如：

• 數(shù)據(jù)查詢平臺(tái)的隊(duì)列是“rt_compute_data_query”
• 數(shù)據(jù)質(zhì)量監(jiān)控平臺(tái)的隊(duì)列是“rt_compute_data_monitor”
• 數(shù)據(jù)鏈路追蹤系統(tǒng)的隊(duì)列是“rt_compute_data_link”

這樣，每個(gè)訂閱這份數(shù)據(jù)的系統(tǒng)其實(shí)都有一個(gè)屬于自己的隊(duì)列，然后隊(duì)列里被會(huì)被exchange路由進(jìn)去實(shí)時(shí)計(jì)算平臺(tái)生產(chǎn)的所有數(shù)據(jù)。

而且因?yàn)槭嵌鄠€(gè)隊(duì)列的模式，每個(gè)系統(tǒng)都可以部署消費(fèi)者集群來進(jìn)行數(shù)據(jù)的消費(fèi)和處理，非常的方便。

八、整體架構(gòu)圖

最后，給大家來一張大圖，我們?cè)俑鴪D，來捋一捋整個(gè)流程。

如上圖所示，首先，實(shí)時(shí)計(jì)算平臺(tái)會(huì)投遞消息到“rt_compute_data”這個(gè)“exchange”里去，但是他沒指定這個(gè)exchange要路由消息到哪個(gè)隊(duì)列，因?yàn)檫@個(gè)他本身是不知道的。

接著數(shù)據(jù)查詢平臺(tái)、數(shù)據(jù)質(zhì)量監(jiān)控系統(tǒng)、數(shù)據(jù)鏈路追蹤系統(tǒng)，就可以聲明自己的隊(duì)列，都綁定到exchange上去。

因?yàn)閝ueue和exchange的綁定，在這里是要由訂閱數(shù)據(jù)的平臺(tái)自己指定的。而且因?yàn)檫@個(gè)exchange是fanout類型的，他只要接收到了數(shù)據(jù)，就會(huì)路由數(shù)據(jù)到所有綁定到他的隊(duì)列里去，這樣每個(gè)隊(duì)列里都有同樣的一份數(shù)據(jù)，供對(duì)應(yīng)的平臺(tái)來消費(fèi)。

而且針對(duì)每個(gè)平臺(tái)自己的隊(duì)列，自己還可以部署消費(fèi)服務(wù)集群來消費(fèi)自己的一個(gè)隊(duì)列，自己的隊(duì)列里的數(shù)據(jù)還是會(huì)均勻分發(fā)給各個(gè)消費(fèi)服務(wù)實(shí)例來處理，每個(gè)消費(fèi)服務(wù)實(shí)例會(huì)獲取到一部分的數(shù)據(jù)。

大家思考一下，這樣是不是就實(shí)現(xiàn)了不同的系統(tǒng)訂閱一份數(shù)據(jù)的“Pub/Sub”的模型？

當(dāng)然，其實(shí)RabbitMQ還支持各種不同類型的exchange，可以實(shí)現(xiàn)各種復(fù)雜的功能。

后續(xù)我們將會(huì)給大家通過實(shí)際的線上系統(tǒng)架構(gòu)案例，來闡述消息中間件技術(shù)的各種用法。