本文整理自Apache RocketMQ&DataPipeline 在北京舉辦的Meetup
本文主要從三方面展開。首先介紹VIPKID使用MQ的歷史，即如何演化到現(xiàn)在的RocketMQ Kafka。其次，為RocketMQ在VIPKID的實踐，主要介紹如何解決在使用RocketMQ過程中出現(xiàn)的相關(guān)問題。最后，提到VIPKID的MQ服務(wù)關(guān)于以后可能會如何變化和發(fā)展。
MQ History in VIPKID
關(guān)于VIPKID的需求：
第一性能。VIPKID是貫穿北美和中國的線上英語學(xué)習(xí)機(jī)構(gòu)，數(shù)據(jù)、消息量龐大，對于MQ吞吐量、延遲等有極度的要求。
第二海量的topic。因為業(yè)務(wù)方的業(yè)務(wù)場景相對比較復(fù)雜，不同于常規(guī)，所以對多topic的要求較高。目前在測試環(huán)節(jié)，包含的topic應(yīng)該有3000 ，消費(fèi)者有超過10k 。
第三統(tǒng)一的管理平臺。由于各種topic，消費(fèi)者紛繁復(fù)雜， 需要一個像Operator那樣可以在平臺上“指指畫畫”即可管理topic，消費(fèi)者，和相關(guān)負(fù)責(zé)人的信息的平臺。
第四友好的API。對RD同學(xué)而言，不需花費(fèi)過多的精力學(xué)習(xí)，只需看API便能上手接入。
另外對提供MQ服務(wù)的同學(xué)而言，方便二次開發(fā)。中間件大部分選擇用Java，如果使用其它語言，比如GO，Scala，對二次開發(fā)的難度會比較大，維護(hù)成本也會很高。
考慮到學(xué)習(xí)成本，以上為我們在之前面臨的一些問題和訴求，基于這些訴求，我們在之前就開始做選擇。
公司之前使用MQ服務(wù)時，管理簡單。這也讓我們面臨一些較為尷尬的問題。
之后經(jīng)過層層篩選，各種把關(guān)，選擇了RocketMQ＋Kafka，RocketMQ主要提供給業(yè)務(wù)方使用，做一些業(yè)務(wù)數(shù)據(jù)的流轉(zhuǎn)。包括像削峰填谷，或是一些異步數(shù)據(jù)等同步，另外還需同步一些BI數(shù)據(jù)給業(yè)務(wù)方。
為什么會這么選？現(xiàn)在市面上有許多比較成熟的MQ產(chǎn)品，比如像目前使用的Kafka，第二個是RocketMQ，第三個為RabbitMQ，最后決定Kafka先保持現(xiàn)狀，把業(yè)務(wù)型的ActiveMQ替換成RocketMQ。
同時，也積極和RocketMQ社區(qū)保持聯(lián)系，一來可以依靠社區(qū)力量將公司MQ服務(wù)做得更加穩(wěn)定。另外也可以向社區(qū)其他同學(xué)分享我們的實踐經(jīng)驗。
為什么Kafka會保持，原因在于對于ELK生態(tài)而言，Kafka目前是不可或缺的，在性能上可以滿足要求。
為什么選擇RocketMQ，有以下原因：低延遲、百萬級topic、二次開發(fā)容易、管理部署方便。
下圖為系統(tǒng)的簡單結(jié)構(gòu)圖，大概描述了生產(chǎn)消費(fèi)的大概過程。還有Broker端的部署結(jié)構(gòu)，最上面一層為生產(chǎn)過程，中間生產(chǎn)到了Broker的集群，下面是一個消費(fèi)者的過程，最右邊為retry的隊列，主要做容錯。
比如發(fā)一個消息時不是一錘子買賣，當(dāng)消息發(fā)過去之后，由于網(wǎng)絡(luò)抖了一下，接下來第二第三API失敗了，此時可不可以允許我再發(fā)一次消息，像這種容錯的情況是支持的，所以RocketMQ也是很重要的原因。
RocketMQ 最佳實踐
首先最重要的問題，也是研發(fā)同學(xué)反饋過來的問題中最難理解的一部分，即訂閱關(guān)系一致。
所謂的訂閱關(guān)系一致，指一個消費(fèi)者組在一個服務(wù)中，假設(shè)消費(fèi)者組消費(fèi)了3個topic，該服務(wù)部署到任何一臺機(jī)器上，消費(fèi)者組消費(fèi)到topic都必須一樣，這就是訂閱關(guān)系一致。因為測試環(huán)節(jié)是多環(huán)境的，在多環(huán)境情況下會遇到訂閱關(guān)系不一致的情況，這屬于使用問題。
第二個是Docker，因為多環(huán)境情況下，在測試環(huán)節(jié)走的是K8S，它會啟動Docker容器，在Docker里會面臨一個問題，消費(fèi)者客戶端在起寫時，會將自己的訂閱信息上報給Broker，告訴它訂閱了哪些topic，以及消費(fèi)者客戶端ID是多少，以此來表明身份。當(dāng)在取客戶端ID時，取了一個本地IP，但是在Docker里面取出來的本地大部分是相同的。
上述情況會出現(xiàn)類似于訂閱關(guān)系不一致的結(jié)果。會發(fā)現(xiàn)topic里面有的消息、隊列沒有消費(fèi)者去訂閱，實際是有的。因為Broker在做rebalance時，由于沒有辦法正常區(qū)分到底起了多少個消費(fèi)者時令，導(dǎo)致分配隊列時出現(xiàn)問題。原因在于當(dāng)上報訂閱關(guān)系給Broker時，數(shù)據(jù)有錯誤，這是Broker從本地IP取出來導(dǎo)致的。
第三個是多環(huán)境問題，假設(shè)有ABC三個服務(wù)，A服務(wù)依賴于B，B依賴于C。通常假設(shè)在開發(fā)一個服務(wù)時，會單獨(dú)拉一個分支進(jìn)行開發(fā)，此時可能會與上一步的下一個服務(wù)做聯(lián)調(diào)。
同時在測試環(huán)節(jié)，還會有一個穩(wěn)定的服務(wù)提供給其他人做聯(lián)調(diào)，在沒有開發(fā)時需要給QA做集成測試。這個環(huán)境中稱為多環(huán)境，在多環(huán)境的情況就會遇到訂閱關(guān)系不一致的問題。同時，在多環(huán)境下如何做消息路由也是需要考慮的問題。
首先解釋訂閱關(guān)系不一致，假設(shè)有兩個人開發(fā)兩個功能，其中一個是Host為1.1.1.1，消費(fèi)者組有3個topic，1、2、3，當(dāng)拿到一個新的需求后需要將topic3換成topic4，獲取消息。
在消費(fèi)寫完之后，需要發(fā)到一個環(huán)境進(jìn)行測試，假設(shè)將這兩個同時放到環(huán)境中，就會出現(xiàn)訂閱關(guān)系不一致，這時有可能會出現(xiàn)以下情況：
當(dāng)添加topic4去掉topic3之后，在測試topic4是否能收到消息時，結(jié)果發(fā)現(xiàn)收不到。或是當(dāng)發(fā)送一條消息之后，連續(xù)不斷地收到了100多萬條消息。
如果打開console平臺，發(fā)現(xiàn)有的隊列沒有消費(fèi)者去消費(fèi)，這就是所謂的訂閱關(guān)系不一致。
比如，同樣的一個ConsumerGroup1，第一臺機(jī)器消費(fèi)了3個topic1、2、3，第二臺機(jī)器消費(fèi)了topic1、2、4，這就是訂閱關(guān)系不一致。即同一個消費(fèi)組部署的兩臺實例，消費(fèi)者的topic不一樣，稱為訂閱關(guān)系不一致。
為此，在多環(huán)境的情況下制定了一個簡單規(guī)則。通常發(fā)這種分支時，是在多環(huán)境的子環(huán)境，比如，當(dāng)發(fā)一個新功能時，需要將開發(fā)分支放到A1環(huán)境進(jìn)行測試，這時就有一個簡單的規(guī)則，即不能跨環(huán)境生產(chǎn)消費(fèi)。
假設(shè)一個開發(fā)分支是A1環(huán)鏡的，當(dāng)消費(fèi)消息時，可用A1環(huán)境的consumerGroup去消費(fèi)A1環(huán)境的topic。如果要測試一下收的消息處理邏輯是否正確，就可以去控制臺模擬發(fā)送給A1的topic。
這時就會發(fā)現(xiàn)，自己的開發(fā)分支跟正常的穩(wěn)定環(huán)境完全不影響。多環(huán)境時無需害怕環(huán)境，只需排查當(dāng)前處于哪個環(huán)境，哪些環(huán)境是不需要訂閱的，那就將其去掉，換成所處的環(huán)境即可。
針對訂閱關(guān)系不一致的情況，可以去客戶端中查看訂閱關(guān)系哪里不一致，哪個環(huán)境的topic或消費(fèi)者不應(yīng)該去訂閱，這時容易排查問題。
另外關(guān)于Docker，當(dāng)消費(fèi)者集起來時，該如何告訴Broker起了多少個實例，以此讓Broker來分配，哪些實例消費(fèi)哪些partition或哪些Q。在consumer集起來時，可將訂閱關(guān)系還有能唯一標(biāo)識是我的客戶端ID發(fā)給Broker，讓Broker決定哪個客戶端ID消費(fèi)哪個隊列，進(jìn)而做好分配，這時就能保證正常消費(fèi)。
那如何標(biāo)識我是我？這時有一個字符串，字符串有三個值組成，第一是消費(fèi)者所在機(jī)器的IP，第二是at符號，第三是instance name，即實例名。
之前的配置是一個consumer中的實例名是相同的，如果是Docker環(huán)境，取出的IP也是相同的。假設(shè)有10臺機(jī)器在Docker里集起來，當(dāng)將消息匯報上去之后，Broker在拿到ID去重后發(fā)現(xiàn)只有一臺機(jī)器啟起來，它便將所有的隊列只分配給這一臺機(jī)器，但這臺機(jī)器很隨機(jī)，其它機(jī)器便收不到消息。此時本來有10臺機(jī)器，結(jié)果只有一臺機(jī)器收到消息，或會重復(fù)收到消息，結(jié)果就會很棘手。
為解決上述問題，在配置時，可去掉消費(fèi)者的instance name，這時它會變成一串?dāng)?shù)字，即當(dāng)前進(jìn)程啟起來的進(jìn)程號。大部分情況，容器中啟起來的進(jìn)程號，我們認(rèn)為是不一樣的，但也有可能是相同的，這時可以重啟一下。但這種概率比較小，如果有兩套集群就有可能了，這樣就解決了Docker的問題。
接下來簡單解釋一下RocketMQ里面的訂閱關(guān)系，一條訂閱關(guān)系指一個消費(fèi)者組訂閱了一個topic的某一個tag，三者合起來稱為一個訂閱關(guān)系。
關(guān)于tag，指topic的一個子分類，topic是對消息邏輯的歸類，我可以為某一些消息打上特定的tag，這樣做的益處是會做一些服務(wù)端過濾，假設(shè)只訂閱某個tag消息，那就不會發(fā)送其它tag消息給你，中間沒有數(shù)據(jù)流量的，這對于網(wǎng)絡(luò)輿論而言會減少很大一部分。
另外，我們做了簡單的MQ云平臺，可以理解為與阿里云平臺是類似的，用戶有租戶隔離，不同的租戶之間看不到彼此的數(shù)據(jù)，所有的管理維度是通過APP的維度。
簡單介紹下，一個應(yīng)用有應(yīng)用的owner，應(yīng)用的member，這些人可以管理APP中的所有資源，包括APP中的生產(chǎn)者組，topic、消費(fèi)者組，通過這些可以完成以下事情：
一位研發(fā)同學(xué)如果想與另外一個團(tuán)隊進(jìn)行消息交互，作為生產(chǎn)消息的人，可以去平臺申請個topic，進(jìn)而發(fā)消息將topic告訴消費(fèi)者。比如，我的topic是這個，你去消費(fèi)吧。在拿到之后，再去創(chuàng)建一個消費(fèi)者組，訂閱這個topic，全程無人干預(yù)，目前需要的干預(yù)是我們會做一層審核。
現(xiàn)在很多人不是很熟悉，所以需要引導(dǎo)一段時間后，再將權(quán)限完全開放出來，屆時審計的工作就給應(yīng)用的owner即可。
應(yīng)用的owner通常是團(tuán)隊的leader，這樣就可以在自己的小組中做到完全自助。對研發(fā)團(tuán)隊而言，可以省去許多繁雜流程。
目前整套MQ的工作過程是如果需要資源，先去VKMQ平臺申請資源，比如我是生產(chǎn)者，這時需要一個topic，一個生產(chǎn)者組來生產(chǎn)消息，此時可以在自己創(chuàng)建之后用平臺客戶端發(fā)，無需擔(dān)心到底需要發(fā)到哪個Broker，哪個name server。
這時研發(fā)同學(xué)用起來會比較方便，不用關(guān)心需要哪些集群。如果沒有這樣的發(fā)現(xiàn)機(jī)制，可能還需要寫B(tài)roker地址或name server地址。
How about the future
提到以后可能的發(fā)展走向，之前與阿里中間件技術(shù)團(tuán)隊專家王昕聊過這個問題，了解到會有OpenMessaging，OpenMessaging的目的是有一套公開的標(biāo)準(zhǔn)，是MQ層面的標(biāo)準(zhǔn)，會定義一些基本操作。就現(xiàn)在情況而言，個人覺得，這三者可以結(jié)合到一起。
目前我們有用Kafka和RocketMQ。這時該如何管理，如何統(tǒng)一，對人的要求就很高。
但具體處理問題時會有差異，我可以大膽猜測一下，可以將兩個可能合而為一，依循OpenMessaging的標(biāo)準(zhǔn)。既然有一套比較好的標(biāo)準(zhǔn)可以遵循，如果能實現(xiàn)統(tǒng)一，我覺得對后來的人就會很方便。
