一份互聯(lián)網(wǎng)大廠的消息中間件生產(chǎn)技術(shù)方案總結(jié),快收藏
1、前情回顧
上一篇文章:《??給你一份精心設(shè)計(jì)的消息中間件高擴(kuò)展架構(gòu),趕緊寫進(jìn)簡歷吧!??》,我們分析了如何利用消息中間件對兩系統(tǒng)進(jìn)行解耦處理。
同時(shí),我們也提到了,使用消息中間件還有利于一份數(shù)據(jù)被多個(gè)系統(tǒng)同時(shí)訂閱,供多個(gè)系統(tǒng)用于不同目的。
目前的一個(gè)架構(gòu)如下圖所示。
在這個(gè)圖里,我們可以清晰的看到,實(shí)時(shí)計(jì)算平臺發(fā)布的一份數(shù)據(jù)到消息中間件里,然后接下來:
- 數(shù)據(jù)查詢平臺會訂閱這份數(shù)據(jù),并落入自己本地的數(shù)據(jù)庫集群和緩存集群里,接著對外提供數(shù)據(jù)查詢的服務(wù)
- 數(shù)據(jù)質(zhì)量監(jiān)控系統(tǒng)會對計(jì)算結(jié)果按照一定的業(yè)務(wù)規(guī)則進(jìn)行監(jiān)控,如果發(fā)現(xiàn)有數(shù)據(jù)計(jì)算錯(cuò)誤,則會立馬進(jìn)行報(bào)警
- 數(shù)據(jù)鏈路追蹤系統(tǒng)會采集計(jì)算結(jié)果作為一個(gè)鏈路節(jié)點(diǎn),同時(shí)對一條數(shù)據(jù)的整個(gè)計(jì)算鏈路都進(jìn)行采集并組裝出來一系列的數(shù)據(jù)計(jì)算鏈路落地存儲,最后如果某個(gè)數(shù)據(jù)計(jì)算錯(cuò)誤了,就可以立馬通過計(jì)算鏈路進(jìn)行回溯排查問題
通過以上回顧,我們已經(jīng)清楚,在上述場景中,使用消息中間件一來可以解耦,二來可以實(shí)現(xiàn)消息“Pub/Sub”模型,實(shí)現(xiàn)消息的發(fā)布與訂閱。
這篇文章,咱們就來落地實(shí)踐一把,基于RabbitMQ消息中間件,如何實(shí)現(xiàn)一份數(shù)據(jù)被多個(gè)系統(tǒng)同時(shí)訂閱的“Pub/Sub”模型?
2、基于消息中間件的隊(duì)列消費(fèi)模型
上圖其實(shí)就是采用的RabbitMQ最基本的隊(duì)列消費(fèi)模型的支持,你可以理解為RabbitMQ內(nèi)部有一個(gè)隊(duì)列,生產(chǎn)者不斷的發(fā)送數(shù)據(jù)到隊(duì)列里,消息按照先后順序進(jìn)入隊(duì)列中排隊(duì)。
現(xiàn)在假設(shè)隊(duì)列里有4條數(shù)據(jù),我們有2個(gè)消費(fèi)者一起消費(fèi)這個(gè)隊(duì)列的數(shù)據(jù)。
此時(shí)每個(gè)消費(fèi)會均勻的分配到2條數(shù)據(jù),也就是說4條數(shù)據(jù)會均勻的分配給各個(gè)消費(fèi)者,每個(gè)消費(fèi)者只不過是處理一部分?jǐn)?shù)據(jù)罷了,這個(gè)就是典型的隊(duì)列消費(fèi)模型。
ps:如果有同學(xué)對基于RabbitMQ如何實(shí)現(xiàn)還不太清楚,可以參考之前的一些文章:
《RabbitMQ是如何收發(fā)消息的?(通俗易懂)》
《車禍現(xiàn)場!線上突然宕機(jī),一條訂單消息丟失了...》
《RabbitMQ宕機(jī)后,消息100%不會丟失嗎》
這幾篇文章給出了上述那個(gè)最基本的隊(duì)列消費(fèi)模型的RabbitMQ代碼實(shí)現(xiàn),以及如何保證消費(fèi)者宕機(jī)時(shí)數(shù)據(jù)不丟失,如何讓RabbitMQ集群對queue和message都進(jìn)行持久化,整體代碼實(shí)現(xiàn)比較完整,可以參考一下。
3、基于消息中間件的“Pub/Sub”模型
除了上述的基本模型外,消息中間件還可以實(shí)現(xiàn)一種“Pub/Sub”模型,也就是“發(fā)布/訂閱”模型,Pub就是Publish,Sub就是Subscribe。
這種模型可以支持多個(gè)系統(tǒng)同時(shí)消費(fèi)一份數(shù)據(jù),也就是說你發(fā)布出去的每條數(shù)據(jù),都會廣播給每個(gè)系統(tǒng),看下圖:
也就是說,我們想要實(shí)現(xiàn)的上圖的效果:實(shí)時(shí)計(jì)算平臺發(fā)布一系列的數(shù)據(jù)到消息中間件里,然后數(shù)據(jù)查詢平臺、數(shù)據(jù)質(zhì)量監(jiān)控系統(tǒng)、數(shù)據(jù)鏈路追蹤系統(tǒng),都會訂閱數(shù)據(jù),都會消費(fèi)到同一份完整的數(shù)據(jù),每個(gè)系統(tǒng)都可以根據(jù)自己的需要使用數(shù)據(jù)。
那么這個(gè)所謂的“Pub/Sub”模型,基于RabbitMQ應(yīng)該怎么來處理呢?
4、RabbitMQ中的exchange到底是個(gè)什么東西?
實(shí)際上,在RabbitMQ里面是不允許生產(chǎn)者直接投遞消息到某個(gè)queue(隊(duì)列)里的,而是只能讓生產(chǎn)者投遞消息給RabbitMQ內(nèi)部的一個(gè)特殊組件,叫做“exchange”,你大概可以理解為一種消息路由組件。
也就是說,實(shí)時(shí)計(jì)算平臺發(fā)送出去的message到RabbitMQ中都是由一個(gè)exchange來接收的。
然后這個(gè)exchange會根據(jù)一定的規(guī)則決定要將這個(gè)message路由轉(zhuǎn)發(fā)到哪個(gè)queue里去,這實(shí)際上就是RabbitMQ中的一個(gè)核心的消息模型。
大家看下面的圖來理解一下。
5、默認(rèn)的exchange
你也許會說,我投遞消息到RabbitMQ的時(shí)候,也沒有用什么exchange,但是為什么還是把消息投遞到了queue里去呢?
那是因?yàn)槟闶褂昧四J(rèn)的exchange,他會直接把消息路由到你指定的那個(gè)queue里去,所以如果簡單用隊(duì)列消費(fèi)模型,就省去了exchange的概念。
上面這段就是之前給大家展示的,讓消息持久化的一種投遞消息的方式。
大家注意里面的第一個(gè)參數(shù),是一個(gè)空的字符串,這個(gè)空字符串的意思,就是說投遞消息到默認(rèn)的exchange里去,然后他就會路由消息到我們指定的queue里去。
6、將消息投遞到fanout exchange
在RabbitMQ里,exchange這種組件有很多種類型,比如說:direct、topic、headers以及fanout,本文我們來看最后一種fanout。
這種exchange組件其實(shí)非常的簡單,你可以創(chuàng)建一個(gè)fanout類型的exchange,然后給這個(gè)exchange綁定多個(gè)queue,接著只要你投遞一條消息到這個(gè)exchange,他就會把消息路由給他綁定的所有queue。
使用下面的代碼就可以創(chuàng)建一個(gè)exchange,比如說在實(shí)時(shí)計(jì)算平臺(生產(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組件里去:
大家會注意到,此時(shí)消息就是投遞到指定的exchange里去了,但是路由到哪個(gè)queue里去呢?此時(shí)我們暫時(shí)還沒確定,要讓消費(fèi)者自己把自己的queue綁定到這個(gè)exchange上去才可以。
7、綁定自己的隊(duì)列到exchange上
對消費(fèi)者的代碼也進(jìn)行修改,之前我們在這里關(guān)閉了autoAck機(jī)制,然后每次都是自己手動(dòng)ack。
上面的代碼里,每個(gè)消費(fèi)者系統(tǒng),都會有一些不一樣,就是每個(gè)消費(fèi)者都需要定義自己的隊(duì)列,然后綁定到exchange上去。
比如說數(shù)據(jù)查詢平臺的隊(duì)列是“rt_compute_data_query”,數(shù)據(jù)質(zhì)量監(jiān)控平臺的隊(duì)列是“rt_compute_data_monitor”,數(shù)據(jù)鏈路追蹤系統(tǒng)的隊(duì)列是“rt_compute_data_link”。
這樣每個(gè)訂閱這份數(shù)據(jù)的系統(tǒng)其實(shí)都有一個(gè)屬于自己的隊(duì)列,然后隊(duì)列里被會被exchange路由進(jìn)去實(shí)時(shí)計(jì)算平臺生產(chǎn)的所有數(shù)據(jù)。
而且因?yàn)槭嵌鄠€(gè)隊(duì)列的模式,每個(gè)系統(tǒng)都可以部署消費(fèi)者集群來進(jìn)行數(shù)據(jù)的消費(fèi)和處理,非常的方便。
8、整體架構(gòu)圖
如上圖所示,實(shí)時(shí)計(jì)算平臺會投遞消息到“rt_compute_data”這個(gè)“exchange”里去,但是他沒指定這個(gè)exchange要路由消息到哪個(gè)隊(duì)列,因?yàn)檫@個(gè)他本身是不知道的。
接著數(shù)據(jù)查詢平臺、數(shù)據(jù)質(zhì)量監(jiān)控系統(tǒng)、數(shù)據(jù)鏈路追蹤系統(tǒng),就可以聲明自己的隊(duì)列,都綁定到exchange上去。
因?yàn)閝ueue和exchange的綁定,在這里是由要訂閱數(shù)據(jù)的平臺自己指定的。而且因?yàn)檫@個(gè)exchange是fanout類型的,他只要接收到了數(shù)據(jù),就會路由數(shù)據(jù)到所有綁定到他的隊(duì)列里去,這樣每個(gè)隊(duì)列里都有同樣的一份數(shù)據(jù),供對應(yīng)的平臺來消費(fèi)。
而且針對每個(gè)平臺自己的隊(duì)列,自己還可以部署消費(fèi)服務(wù)集群來消費(fèi)自己的一個(gè)隊(duì)列,自己的隊(duì)列里的數(shù)據(jù)還是會均勻分發(fā)給各個(gè)消費(fèi)服務(wù)實(shí)例來處理,每個(gè)消費(fèi)服務(wù)實(shí)例會獲取到一部分的數(shù)據(jù)。
這樣是不是就實(shí)現(xiàn)了不同的系統(tǒng)訂閱一份數(shù)據(jù)的“Pub/Sub”的模型?
當(dāng)然,RabbitMQ還支持各種不同類型的exchange,可以實(shí)現(xiàn)各種復(fù)雜的功能,后續(xù)我們再來給大家通過實(shí)際的線上系統(tǒng)架構(gòu)案例,來闡述消息中間件技術(shù)的用法。?