日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

【稿件】上周分享的一篇《面試大殺器：為什么一定要用MQ中間件?》受到了大家的一致好評(píng)，今天這篇文章為大家總結(jié)下 MQ 應(yīng)用中的一些疑難雜癥。

創(chuàng)新互聯(lián)公司是由多位在大型網(wǎng)絡(luò)公司、廣告設(shè)計(jì)公司的優(yōu)秀設(shè)計(jì)人員和策劃人員組成的一個(gè)具有豐富經(jīng)驗(yàn)的團(tuán)隊(duì)，其中包括網(wǎng)站策劃、網(wǎng)頁(yè)美工、網(wǎng)站程序員、網(wǎng)頁(yè)設(shè)計(jì)師、平面廣告設(shè)計(jì)師、網(wǎng)絡(luò)營(yíng)銷人員及形象策劃。承接：成都網(wǎng)站制作、成都網(wǎng)站建設(shè)、網(wǎng)站改版、網(wǎng)頁(yè)設(shè)計(jì)制作、網(wǎng)站建設(shè)與維護(hù)、網(wǎng)絡(luò)推廣、數(shù)據(jù)庫(kù)開發(fā),以高性價(jià)比制作企業(yè)網(wǎng)站、行業(yè)門戶平臺(tái)等全方位的服務(wù)。

消息隊(duì)列有什么優(yōu)點(diǎn)和缺點(diǎn)?

為什么使用消息隊(duì)列?假設(shè)你的業(yè)務(wù)場(chǎng)景遇到個(gè)技術(shù)挑戰(zhàn)，如果不用 MQ 可能會(huì)很麻煩，但是你用了 MQ 之后會(huì)帶給你很多好處。

消息隊(duì)列 MQ 的常見(jiàn)使用場(chǎng)景其實(shí)有很多，但是比較核心的有如下三個(gè)：

• 解耦
• 異步
• 削峰

解耦：A 系統(tǒng)發(fā)送個(gè)數(shù)據(jù)到 BCD 三個(gè)系統(tǒng)，接口調(diào)用發(fā)送，那如果 E 系統(tǒng)也要這個(gè)數(shù)據(jù)呢?那如果 C 系統(tǒng)現(xiàn)在不需要了呢?

現(xiàn)在 A 系統(tǒng)又要發(fā)送第二種數(shù)據(jù)了呢?而且 A 系統(tǒng)要時(shí)時(shí)刻刻考慮 BCDE 四個(gè)系統(tǒng)如果掛了咋辦?要不要重發(fā)?我要不要把消息存起來(lái)?

你需要去考慮一下你負(fù)責(zé)的系統(tǒng)中是否有類似的場(chǎng)景，就是一個(gè)系統(tǒng)或者一個(gè)模塊，調(diào)用了多個(gè)系統(tǒng)或者模塊，互相之間的調(diào)用很復(fù)雜，維護(hù)起來(lái)很麻煩。

但是，這個(gè)調(diào)用是不需要直接同步調(diào)用接口的，如果用 MQ 給他異步化解耦，也是可以的。你就只需要去考慮在你的項(xiàng)目里，是不是可以運(yùn)用這個(gè) MQ 去進(jìn)行系統(tǒng)的解耦。

異步：A 系統(tǒng)接收一個(gè)請(qǐng)求，需要在自己本地寫庫(kù)，還需要在 BCD 三個(gè)系統(tǒng)寫庫(kù)，自己本地寫庫(kù)要 30ms，BCD 三個(gè)系統(tǒng)分別寫庫(kù)要 300ms、450ms、200ms。

最終請(qǐng)求總延時(shí)是 30 + 300 + 450 + 200 = 980ms，接近 1s，異步后，BCD 三個(gè)系統(tǒng)分別寫庫(kù)的時(shí)間，A 系統(tǒng)就不再考慮了。

削峰：每天 0 點(diǎn)到 16 點(diǎn)，A 系統(tǒng)風(fēng)平浪靜，每秒并發(fā)請(qǐng)求數(shù)量就 100 個(gè)。結(jié)果每次一到 16 點(diǎn)~23 點(diǎn)，每秒并發(fā)請(qǐng)求數(shù)量突然會(huì)暴增到 10000 條。

但是系統(tǒng)的處理能力就只能是每秒鐘處理 1000 個(gè)請(qǐng)求啊。怎么辦?需要我們進(jìn)行流量的削峰，讓系統(tǒng)可以平緩的處理突增的請(qǐng)求。

優(yōu)點(diǎn)上面已經(jīng)說(shuō)了，就是在特殊場(chǎng)景下有其對(duì)應(yīng)的好處，解耦、異步、削峰，那么消息隊(duì)列有什么缺點(diǎn)?

系統(tǒng)可用性降低：系統(tǒng)引入的外部依賴越多，越容易掛掉，本來(lái)你就是 A 系統(tǒng)調(diào)用 BCD 三個(gè)系統(tǒng)的接口就好了。

ABCD 四個(gè)系統(tǒng)好好的，沒(méi)啥問(wèn)題，你偏加個(gè) MQ 進(jìn)來(lái)，萬(wàn)一 MQ 掛了怎么辦?MQ 掛了，整套系統(tǒng)崩潰了，業(yè)務(wù)也就停頓了。

系統(tǒng)復(fù)雜性提高：硬生生加個(gè) MQ 進(jìn)來(lái)，怎么保證消息沒(méi)有重復(fù)消費(fèi)?怎么處理消息丟失的情況?怎么保證消息傳遞的順序性?

一致性問(wèn)題：A 系統(tǒng)處理完了直接返回成功了，大家都以為你這個(gè)請(qǐng)求就成功了。

但問(wèn)題是，要是 BCD 三個(gè)系統(tǒng)那里，BD 兩個(gè)系統(tǒng)寫庫(kù)成功了，結(jié)果 C 系統(tǒng)寫庫(kù)失敗了，你這數(shù)據(jù)就不一致了。

所以消息隊(duì)列實(shí)際是一種非常復(fù)雜的架構(gòu)，你引入它有很多好處，但是也得針對(duì)它帶來(lái)的壞處做各種額外的技術(shù)方案和架構(gòu)來(lái)規(guī)避掉。

常見(jiàn)消息隊(duì)列的比較如下圖：

如何解決重復(fù)消費(fèi)?

消息重復(fù)的原因

消息發(fā)送端應(yīng)用的消息重復(fù)發(fā)送，有以下幾種情況：

• 消息發(fā)送端發(fā)送消息給消息中間件，消息中間件收到消息并成功存儲(chǔ)，而這時(shí)消息中間件出現(xiàn)了問(wèn)題，導(dǎo)致應(yīng)用端沒(méi)有收到消息發(fā)送成功的返回因而進(jìn)行重試產(chǎn)生了重復(fù)。
• 消息中間件因?yàn)樨?fù)載高響應(yīng)變慢，成功把消息存儲(chǔ)到消息存儲(chǔ)中后，返回“成功”這個(gè)結(jié)果時(shí)超時(shí)。
• 消息中間件將消息成功寫入消息存儲(chǔ)，在返回結(jié)果時(shí)網(wǎng)絡(luò)出現(xiàn)問(wèn)題，導(dǎo)致應(yīng)用發(fā)送端重試，而重試時(shí)網(wǎng)絡(luò)恢復(fù)，由此導(dǎo)致重復(fù)。

可以看到，通過(guò)消息發(fā)送端產(chǎn)生消息重復(fù)的主要原因是消息成功進(jìn)入消息存儲(chǔ)后，因?yàn)楦鞣N原因使得消息發(fā)送端沒(méi)有收到“成功”的返回結(jié)果，并且又有重試機(jī)制，因而導(dǎo)致重復(fù)。

消息到達(dá)了消息存儲(chǔ)，由消息中間件進(jìn)行向外的投遞時(shí)產(chǎn)生重復(fù)，有以下幾種情況：

• 消息被投遞到消息接收者應(yīng)用進(jìn)行處理，處理完畢后應(yīng)用出問(wèn)題了，消息中間件不知道消息處理結(jié)果，會(huì)再次投遞。
• 消息被投遞到消息接收者應(yīng)用進(jìn)行處理，處理完畢后網(wǎng)絡(luò)出現(xiàn)問(wèn)題了，消息中間件沒(méi)有收到消息處理結(jié)果，會(huì)再次投遞。
• 消息被投遞到消息接收者應(yīng)用進(jìn)行處理，處理時(shí)間比較長(zhǎng)，消息中間件因?yàn)橄⒊瑫r(shí)會(huì)再次投遞。
• 消息被投遞到消息接收者應(yīng)用進(jìn)行處理，處理完畢后消息中間件出問(wèn)題了,沒(méi)能收到消息結(jié)果并處理，會(huì)再次投遞。
• 消息被投遞到消息接收者應(yīng)用進(jìn)行處理，處理完畢后消息中間件收到結(jié)果但是遇到消息存儲(chǔ)故障，沒(méi)能更新投遞狀態(tài)，會(huì)再次投遞。

可以看到，在投遞過(guò)程中產(chǎn)生的消息重復(fù)接收主要是因?yàn)橄⒔邮照叱晒μ幚硗晗⒑螅⒅虚g件不能及時(shí)更新投遞狀態(tài)造成的。

如何解決重復(fù)消費(fèi)

那么有什么辦法可以解決呢?主要是要求消息接收者來(lái)處理這種重復(fù)的情況，也就是要求消息接收者的消息處理是冪等操作。

什么是冪等性?對(duì)于消息接收端的情況，冪等的含義是采用同樣的輸入多次調(diào)用處理函數(shù)，得到同樣的結(jié)果。

例如，一個(gè) SQL 操作：

 
 
 
 

  
  
  
  
update stat_table set count= 10 where id =1
 
 
 
 

這個(gè)操作多次執(zhí)行，id 等于 1 的記錄中的 count 字段的值都為 10，這個(gè)操作就是冪等的，我們不用擔(dān)心這個(gè)操作被重復(fù)。

再來(lái)看另外一個(gè) SQL 操作：

 
 
 
 

  
  
  
  
update stat_table set count= count +1 where id= 1;
 
 
 
 

這樣的 SQL 操作就不是冪等的，一旦重復(fù)，結(jié)果就會(huì)產(chǎn)生變化。

因此應(yīng)對(duì)消息重復(fù)的辦法是使消息接收端的處理是一個(gè)冪等操作。這樣的做法降低了消息中間件的整體復(fù)雜性，不過(guò)也給使用消息中間件的消息接收端應(yīng)用帶來(lái)了一定的限制和門檻。

①M(fèi)VCC

多版本并發(fā)控制，樂(lè)觀鎖的一種實(shí)現(xiàn)，在生產(chǎn)者發(fā)送消息時(shí)進(jìn)行數(shù)據(jù)更新時(shí)需要帶上數(shù)據(jù)的版本號(hào)，消費(fèi)者去更新時(shí)需要去比較持有數(shù)據(jù)的版本號(hào)，版本號(hào)不一致的操作無(wú)法成功。

例如博客點(diǎn)贊次數(shù)自動(dòng) +1 的接口：

 
 
 
 

  
  
  
  
public boolean addCount(Long id, Long version);
  
  
  
  
update blogTable set count= count+1,version=version+1 where id=321 and version=123
 
 
 
 

每一個(gè) version 只有一次執(zhí)行成功的機(jī)會(huì)，一旦失敗了生產(chǎn)者必須重新獲取數(shù)據(jù)的新版本號(hào)再次發(fā)起更新。

②去重表

利用數(shù)據(jù)庫(kù)表單的特性來(lái)實(shí)現(xiàn)冪等，常用的一個(gè)思路是在表上構(gòu)建索引，保證某一類數(shù)據(jù)一旦執(zhí)行完畢，后續(xù)同樣的請(qǐng)求不再重復(fù)處理了(利用一張日志表來(lái)記錄已經(jīng)處理成功的消息的 id，如果新到的消息 id 已經(jīng)在日志表中，那么就不再處理這條消息。)

以電商平臺(tái)為例子，電商平臺(tái)上的訂單 id 就是最適合的 token。當(dāng)用戶下單時(shí)，會(huì)經(jīng)歷多個(gè)環(huán)節(jié)，比如生成訂單，減庫(kù)存，減優(yōu)惠券等等。

每一個(gè)環(huán)節(jié)執(zhí)行時(shí)都先檢測(cè)一下該訂單 id 是否已經(jīng)執(zhí)行過(guò)這一步驟，對(duì)未執(zhí)行的請(qǐng)求，執(zhí)行操作并緩存結(jié)果，而對(duì)已經(jīng)執(zhí)行過(guò)的 id，則直接返回之前的執(zhí)行結(jié)果，不做任何操作。

這樣可以在更大程度上避免操作的重復(fù)執(zhí)行問(wèn)題，緩存起來(lái)的執(zhí)行結(jié)果也能用于事務(wù)的控制等。

如何保證消息的可靠性傳輸?

ActiveMQ

要保證消息的可靠性，除了消息的持久化，還包括兩個(gè)方面：

• 生產(chǎn)者發(fā)送的消息可以被 ActiveMQ 收到。
• 消費(fèi)者收到了 ActiveMQ 發(fā)送的消息。

①生產(chǎn)者

非持久化又不在事務(wù)中的消息，可能會(huì)有消息的丟失。為保證消息可以被 ActiveMQ 收到，我們應(yīng)該采用事務(wù)消息或持久化消息。

②消費(fèi)者

消費(fèi)者對(duì)消息的確認(rèn)有四種機(jī)制：

• AUTO_ACKNOWLEDGE=1：自動(dòng)確認(rèn)
• CLIENT_ACKNOWLEDGE=2：客戶端手動(dòng)確認(rèn)
• DUPS_OK_ACKNOWLEDGE=3：自動(dòng)批量確認(rèn)
• SESSION_TRANSACTED=0：事務(wù)提交并確認(rèn)

ACK_MODE 描述了 Consumer 與 Broker 確認(rèn)消息的方式(時(shí)機(jī))，比如當(dāng)消息被 Consumer 接收之后，Consumer 將在何時(shí)確認(rèn)消息。

所以 ack_mode 描述的不是 Producer 與 Broker 之間的關(guān)系，而是 Customer 與 Broker 之間的關(guān)系。

對(duì)于 Broker 而言，只有接收到 ACK 指令，才會(huì)認(rèn)為消息被正確的接收或者處理成功了。通過(guò) ACK，可以在 Consumer 與 Broker 之間建立一種簡(jiǎn)單的“擔(dān)保”機(jī)制。

AUTO_ACKNOWLEDGE：自動(dòng)確認(rèn)，“同步”(receive)方法返回 message 給消息時(shí)會(huì)立即確認(rèn)。

在"異步"(messageListener)方式中，將會(huì)首先調(diào)用listener.onMessage(message)。

如果 onMessage 方法正常結(jié)束，消息將會(huì)正常確認(rèn);如果 onMessage 方法異常，將導(dǎo)致消費(fèi)者要求 ActiveMQ 重發(fā)消息。

CLIENT_ACKNOWLEDGE：客戶端手動(dòng)確認(rèn)，這就意味著 AcitveMQ 將不會(huì)“自作主張”的為你 ACK 任何消息，開發(fā)者需要自己擇機(jī)確認(rèn)。

我們可以在當(dāng)前消息處理成功之后，立即調(diào)用 message.acknowledge() 方法來(lái)"逐個(gè)"確認(rèn)消息，這樣可以盡可能的減少因網(wǎng)絡(luò)故障而導(dǎo)致消息重發(fā)的個(gè)數(shù)。

當(dāng)然也可以處理多條消息之后，間歇性的調(diào)用 ACKNOWLEDGE 方法來(lái)一次確認(rèn)多條消息，減少 ACK 的次數(shù)來(lái)提升 Consumer 的效率，不過(guò)需要自行權(quán)衡。

DUPS_OK_ACKNOWLEDGE：類似于 AUTO_ACK 確認(rèn)機(jī)制，為自動(dòng)批量確認(rèn)而生，而且具有“延遲”確認(rèn)的特點(diǎn)，ActiveMQ 會(huì)根據(jù)內(nèi)部算法，在收到一定數(shù)量的消息自動(dòng)進(jìn)行確認(rèn)。

在此模式下，可能會(huì)出現(xiàn)重復(fù)消息，什么時(shí)候?當(dāng) Consumer 故障重啟后，那些尚未 ACK 的消息會(huì)重新發(fā)送過(guò)來(lái)。

SESSION_TRANSACTED：當(dāng) Session 使用事務(wù)時(shí)，就是使用此模式。當(dāng)決定事務(wù)中的消息可以確認(rèn)時(shí)，必須調(diào)用 session.commit() 方法，Commit 方法將會(huì)導(dǎo)致當(dāng)前 Session 的事務(wù)中所有消息立即被確認(rèn)。

在事務(wù)開始之后的任何時(shí)機(jī)調(diào)用 rollback()，意味著當(dāng)前事務(wù)的結(jié)束，事務(wù)中所有的消息都將被重發(fā)。當(dāng)然在 Commit 之前拋出異常，也會(huì)導(dǎo)致事務(wù)的 rollback。

RabbitMQ

①生產(chǎn)者弄丟了數(shù)據(jù)

生產(chǎn)者將數(shù)據(jù)發(fā)送到 RabbitMQ 的時(shí)候，可能數(shù)據(jù)就在半路給搞丟了，因?yàn)榫W(wǎng)絡(luò)啥的問(wèn)題，都有可能。

此時(shí)可以選擇用 RabbitMQ 提供的事務(wù)功能，就是生產(chǎn)者發(fā)送數(shù)據(jù)之前開啟 RabbitMQ 事務(wù)(channel.txSelect)，然后發(fā)送消息，如果消息沒(méi)有成功被 RabbitMQ 接收到，那么生產(chǎn)者會(huì)收到異常報(bào)錯(cuò)。

此時(shí)就可以回滾事務(wù)(channel.txRollback)，然后重試發(fā)送消息;如果收到了消息，那么可以提交事務(wù)(channel.txCommit)。

但是問(wèn)題是，RabbitMQ 事務(wù)機(jī)制一搞，基本上吞吐量會(huì)下來(lái)，因?yàn)樘男阅堋?/p>

所以一般來(lái)說(shuō)，如果要確保 RabbitMQ 的消息別丟，可以開啟 Confirm 模式。

在生產(chǎn)者那里設(shè)置開啟 Confirm 模式之后，你每次寫的消息都會(huì)分配一個(gè) id，然后如果寫入了 RabbitMQ 中，RabbitMQ 會(huì)給你回傳一個(gè) ACK 消息，告訴你說(shuō)這個(gè)消息 OK 了。

如果 RabbitMQ 沒(méi)能處理這個(gè)消息，會(huì)回調(diào)你一個(gè) nack 接口，告訴你這個(gè)消息接收失敗，你可以重試。

而且你可以結(jié)合這個(gè)機(jī)制，自己在內(nèi)存里維護(hù)每個(gè)消息 id 的狀態(tài)，如果超過(guò)一定時(shí)間還沒(méi)接收到這個(gè)消息的回調(diào)，那么你可以重發(fā)。

事務(wù)機(jī)制和 Cnofirm 機(jī)制的不同在于：事務(wù)機(jī)制是同步的，你提交一個(gè)事務(wù)之后會(huì)阻塞在那兒。

但是 Confirm 機(jī)制是異步的，你發(fā)送個(gè)消息之后就可以發(fā)送下一個(gè)消息，然后那個(gè)消息 RabbitMQ 接收了之后會(huì)異步回調(diào)你一個(gè)接口通知你這個(gè)消息接收到了。

所以一般在生產(chǎn)者這塊避免數(shù)據(jù)丟失，都是用 Confirm 機(jī)制的。

②RabbitMQ 弄丟了數(shù)據(jù)

就是 RabbitMQ 自己弄丟了數(shù)據(jù)，這個(gè)你必須開啟 RabbitMQ 的持久化，就是消息寫入之后會(huì)持久化到磁盤，哪怕是 RabbitMQ 自己掛了，恢復(fù)之后會(huì)自動(dòng)讀取之前存儲(chǔ)的數(shù)據(jù)，一般數(shù)據(jù)不會(huì)丟。

除非極其罕見(jiàn)的是，RabbitMQ 還沒(méi)持久化，自己就掛了，可能導(dǎo)致少量數(shù)據(jù)會(huì)丟失的，但是這個(gè)概率較小。

設(shè)置持久化有兩個(gè)步驟：

• 創(chuàng)建 queue 和交換器的時(shí)候?qū)⑵湓O(shè)置為持久化的，這樣就可以保證 RabbitMQ 持久化相關(guān)的元數(shù)據(jù)，但是不會(huì)持久化 queue 里的數(shù)據(jù)。
• 發(fā)送消息的時(shí)候?qū)⑾⒌?deliveryMode 設(shè)置為 2，就是將消息設(shè)置為持久化的，此時(shí) RabbitMQ 就會(huì)將消息持久化到磁盤上去。

必須要同時(shí)設(shè)置這兩個(gè)持久化才行，RabbitMQ 哪怕是掛了，再次重啟，也會(huì)從磁盤上重啟恢復(fù) queue，恢復(fù)這個(gè) queue 里的數(shù)據(jù)。

而且持久化可以跟生產(chǎn)者那邊的 Confirm 機(jī)制配合起來(lái)，只有消息被持久化到磁盤之后，才會(huì)通知生產(chǎn)者 ACK 了。

所以哪怕是在持久化到磁盤之前，RabbitMQ 掛了，數(shù)據(jù)丟了，生產(chǎn)者收不到 ACK，你也是可以自己重發(fā)的。

哪怕是你給 RabbitMQ 開啟了持久化機(jī)制，也有一種可能，就是這個(gè)消息寫到了 RabbitMQ 中，但是還沒(méi)來(lái)得及持久化到磁盤上，結(jié)果不巧，此時(shí) RabbitMQ 掛了，就會(huì)導(dǎo)致內(nèi)存里的一點(diǎn)點(diǎn)數(shù)據(jù)會(huì)丟失。

③消費(fèi)端弄丟了數(shù)據(jù)

RabbitMQ 如果丟失了數(shù)據(jù)，主要是因?yàn)槟阆M(fèi)的時(shí)候，剛消費(fèi)到，還沒(méi)處理，結(jié)果進(jìn)程掛了，比如重啟了，那么就尷尬了，RabbitMQ 認(rèn)為你都消費(fèi)了，這數(shù)據(jù)就丟了。

這個(gè)時(shí)候得用 RabbitMQ 提供的 ACK 機(jī)制，簡(jiǎn)單來(lái)說(shuō)，就是你關(guān)閉 RabbitMQ 自動(dòng) ACK，可以通過(guò)一個(gè) API 來(lái)調(diào)用就行，然后每次你自己代碼里確保處理完的時(shí)候，再程序里 ACK 一把。

這樣的話，如果你還沒(méi)處理完，不就沒(méi)有 ACK?那 RabbitMQ 就認(rèn)為你還沒(méi)處理完，這個(gè)時(shí)候 RabbitMQ 會(huì)把這個(gè)消費(fèi)分配給別的 Consumer 去處理，消息是不會(huì)丟的。

Kafka

①消費(fèi)端弄丟了數(shù)據(jù)

只有一個(gè)可能導(dǎo)致消費(fèi)者弄丟數(shù)據(jù)的情況，就是說(shuō)，你那個(gè)消費(fèi)到了這個(gè)消息，然后消費(fèi)者那邊自動(dòng)提交了 Offset，讓 Kafka 以為你已經(jīng)消費(fèi)好了這個(gè)消息。

其實(shí)你剛準(zhǔn)備處理這個(gè)消息，你還沒(méi)處理，你自己就掛了，此時(shí)這條消息就丟咯。

大家都知道 Kafka 會(huì)自動(dòng)提交 Offset，那么只要關(guān)閉自動(dòng)提交 Offset，在處理完之后自己手動(dòng)提交 Offset，就可以保證數(shù)據(jù)不會(huì)丟。

但是此時(shí)確實(shí)還是會(huì)重復(fù)消費(fèi)，比如你剛處理完，還沒(méi)提交 Offset，結(jié)果自己掛了，此時(shí)肯定會(huì)重復(fù)消費(fèi)一次，自己保證冪等性就好了。

生產(chǎn)環(huán)境碰到的一個(gè)問(wèn)題，就是說(shuō)我們的 Kafka 消費(fèi)者消費(fèi)到了數(shù)據(jù)之后是寫到一個(gè)內(nèi)存的 queue 里先緩沖一下，結(jié)果有的時(shí)候，你剛把消息寫入內(nèi)存 queue，然后消費(fèi)者會(huì)自動(dòng)提交 Offset。

然后此時(shí)我們重啟了系統(tǒng)，就會(huì)導(dǎo)致內(nèi)存 queue 里還沒(méi)來(lái)得及處理的數(shù)據(jù)就丟失了。

②Kafka 弄丟了數(shù)據(jù)

這塊比較常見(jiàn)的一個(gè)場(chǎng)景，就是 Kafka 某個(gè) Broker 宕機(jī)，然后重新選舉 Partition 的 Leader 時(shí)。

大家想想，要是此時(shí)其他的 Follower 剛好還有些數(shù)據(jù)沒(méi)有同步，結(jié)果此時(shí) Leader 掛了，然后選舉某個(gè) Follower 成 Leader 之后，他不就少了一些數(shù)據(jù)?這就丟了一些數(shù)據(jù)啊。

所以此時(shí)一般是要求起碼設(shè)置如下四個(gè)參數(shù)：

• 給這個(gè) Topic 設(shè)置 replication.factor 參數(shù)：這個(gè)值必須大于 1，要求每個(gè) Partition 必須有至少 2 個(gè)副本。
• 在 Kafka 服務(wù)端設(shè)置 min.insync.replicas 參數(shù)：這個(gè)值必須大于 1，這個(gè)是要求一個(gè) Leader 至少感知到有至少一個(gè) Follower 還跟自己保持聯(lián)系，沒(méi)掉隊(duì)，這樣才能確保 Leader 掛了還有一個(gè) Follower 吧。
• 在 Producer 端設(shè)置 acks=all：這個(gè)是要求每條數(shù)據(jù)，必須是寫入所有 Replica 之后，才能認(rèn)為是寫成功了。
• 在 Producer 端設(shè)置 retries=MAX(很大很大很大的一個(gè)值，反復(fù)重試的意思)：這個(gè)是要求一旦寫入失敗，就循環(huán)重試，卡在這里了。

③生產(chǎn)者會(huì)不會(huì)弄丟數(shù)據(jù)

如果按照上述的思路設(shè)置了 ack=all，一定不會(huì)丟，要求是，你的 Leader 接收到消息，所有的 Follower 都同步到了消息之后，才認(rèn)為本次寫成功了。如果沒(méi)滿足這個(gè)條件，生產(chǎn)者會(huì)自動(dòng)不斷的重試，重試無(wú)數(shù)次。

消息的順序性

從根本上說(shuō)，異步消息是不應(yīng)該有順序依賴的，在 MQ 上估計(jì)是沒(méi)法解決。

要實(shí)現(xiàn)嚴(yán)格的順序消息，簡(jiǎn)單且可行的辦法就是：保證生產(chǎn)者、MQServer、消費(fèi)者是一對(duì)一對(duì)一的關(guān)系。

ActiveMQ

①通過(guò)高級(jí)特性 Consumer 獨(dú)有消費(fèi)者(exclusive consumer)

 
 
 
 

  
  
  
  
queue = new ActiveMQQueue("TEST.QUEUE?consumer.exclusive=true");
  
  
  
  
consumer = session.createConsumer(queue);
 
 
 
 

當(dāng)在接收信息的時(shí)候，有多個(gè)獨(dú)占消費(fèi)者的時(shí)候，只有一個(gè)獨(dú)占消費(fèi)者可以接收到消息。

獨(dú)占消息就是在有多個(gè)消費(fèi)者同時(shí)消費(fèi)一個(gè) queue 時(shí)，可以保證只有一個(gè)消費(fèi)者可以消費(fèi)消息。

這樣雖然保證了消息的順序問(wèn)題，不過(guò)也帶來(lái)了一個(gè)問(wèn)題，就是這個(gè) queue 的所有消息將只會(huì)在這一個(gè)主消費(fèi)者上消費(fèi)，其他消費(fèi)者將閑置，達(dá)不到負(fù)載均衡分配。

而實(shí)際業(yè)務(wù)我們可能更多的是這樣的場(chǎng)景，比如一個(gè)訂單會(huì)發(fā)出一組順序消息，我們只要求這一組消息是順序消費(fèi)的，而訂單與訂單之間又是可以并行消費(fèi)的，不需要順序，因?yàn)轫樞蛞矝](méi)有任何意義。

有沒(méi)有辦法做到呢?可以利用 ActiveMQ 的另一個(gè)高級(jí)特性之 messageGroup。

②利用 ActiveMQ 的高級(jí)特性：Message Groups

Message Groups 特性是一種負(fù)載均衡的機(jī)制。在一個(gè)消息被分發(fā)到 Consumer 之前，Broker 首先檢查消息 JMSXGroupID 屬性。

如果存在，那么 Broker 會(huì)檢查是否有某個(gè) Consumer 擁有這個(gè) Message Group。

如果沒(méi)有，那么 Broker 會(huì)選擇一個(gè) Consumer，并將它關(guān)聯(lián)到這個(gè) Message Group。

此后，這個(gè) Consumer 會(huì)接收這個(gè) Message Group 的所有消息，直到 Consumer 被關(guān)閉。

Message Group 被關(guān)閉，通過(guò)發(fā)送一個(gè)消息，并設(shè)置這個(gè)消息的 JMSXGroupSeq 為 -1。

 
 
 
 

  
  
  
  
bytesMessage.setStringProperty("JMSXGroupID", "constact-20100000002");
  
  
  
  
bytesMessage.setIntProperty("JMSXGroupSeq", -1);
 
 
 
 

如上圖所示，同一個(gè) queue 中，擁有相同 JMSXGroupID 的消息將發(fā)往同一個(gè)消費(fèi)者，解決順序問(wèn)題;不同分組的消息又能被其他消費(fèi)者并行消費(fèi)，解決負(fù)載均衡的問(wèn)題。

RabbitMQ

如果有順序依賴的消息，要保證消息有一個(gè) hashKey，類似于數(shù)據(jù)庫(kù)表分區(qū)的的分區(qū) key 列。保證對(duì)同一個(gè) key 的消息發(fā)送到相同的隊(duì)列。

A 用戶產(chǎn)生的消息(包括創(chuàng)建消息和刪除消息)都按 A 的 hashKey 分發(fā)到同一個(gè)隊(duì)列。

只需要把強(qiáng)相關(guān)的兩條消息基于相同的路由就行了，也就是說(shuō)經(jīng)過(guò) m1 和 m2 的在路由表里的路由是一樣的，那自然 m1 會(huì)優(yōu)先于 m2 去投遞。而且一個(gè) queue 只對(duì)應(yīng)一個(gè) Consumer。

Kafka

一個(gè) Topic，一個(gè) Partition，一個(gè) Consumer，內(nèi)部單線程消費(fèi)。

如何解決消息隊(duì)列的延時(shí)以及過(guò)期失效問(wèn)題?RabbitMQ 是可以設(shè)置過(guò)期時(shí)間的，就是 TTL。

如果消息在 queue 中積壓超過(guò)一定的時(shí)間，而又沒(méi)有設(shè)置死信隊(duì)列機(jī)制，就會(huì)被 RabbitMQ 給清理掉，這個(gè)數(shù)據(jù)就沒(méi)了。ActiveMQ 則通過(guò)更改配置，支持消息的定時(shí)發(fā)送。

有幾百萬(wàn)消息持續(xù)積壓幾小時(shí)怎么解決?

發(fā)生了線上故障，幾千萬(wàn)條數(shù)據(jù)在 MQ 里積壓很久。是修復(fù) Consumer 的問(wèn)題，讓他恢復(fù)消費(fèi)速度，然后等待幾個(gè)小時(shí)消費(fèi)完畢?這是個(gè)解決方案，不過(guò)有時(shí)候我們還會(huì)進(jìn)行臨時(shí)緊急擴(kuò)容。

一個(gè)消費(fèi)者一秒是 1000 條，3 個(gè)消費(fèi)者一秒是 3000 條，一分鐘是 18 萬(wàn)條。

所以如果積壓了幾百萬(wàn)到上千萬(wàn)的數(shù)據(jù)，即使消費(fèi)者恢復(fù)了，也需要大概一小時(shí)的時(shí)間才能恢復(fù)過(guò)來(lái)。

一般這個(gè)時(shí)候，只能操作臨時(shí)緊急擴(kuò)容了，具體操作步驟和思路如下：

• 先修復(fù) Consumer 的問(wèn)題，確保其恢復(fù)消費(fèi)速度，然后將現(xiàn)有 Consumer 都停掉。
• 新建一個(gè) Topic，Partition 是原來(lái)的 10 倍，臨時(shí)建立好原先 10 倍或者 20 倍的 queue 數(shù)量。

然后寫一個(gè)臨時(shí)的分發(fā)數(shù)據(jù)的 Consumer 程序，這個(gè)程序部署上去消費(fèi)積壓的數(shù)據(jù)，消費(fèi)之后不做耗時(shí)的處理，直接均勻輪詢寫入臨時(shí)建立好的 10 倍數(shù)量的 queue。

• 接著臨時(shí)征用 10 倍的機(jī)器來(lái)部署 Consumer，每一批 Consumer 消費(fèi)一個(gè)臨時(shí) queue 的數(shù)據(jù)。
• 這種做法相當(dāng)于是臨時(shí)將 queue 資源和 Consumer 資源擴(kuò)大 10 倍，以正常的 10 倍速度來(lái)消費(fèi)數(shù)據(jù)。
• 等快速消費(fèi)完積壓數(shù)據(jù)之后，再恢復(fù)原先部署架構(gòu)，重新用原先的 Consumer 機(jī)器來(lái)消費(fèi)消息。

Kafka是如何實(shí)現(xiàn)高性能的?

①宏觀架構(gòu)層面利用 Partition 實(shí)現(xiàn)并行處理

Kafka 中每個(gè) Topic 都包含一個(gè)或多個(gè) Partition，不同 Partition 可位于不同節(jié)點(diǎn)。

同時(shí) Partition 在物理上對(duì)應(yīng)一個(gè)本地文件夾，每個(gè) Partition 包含一個(gè)或多個(gè) Segment，每個(gè) Segment 包含一個(gè)數(shù)據(jù)文件和一個(gè)與之對(duì)應(yīng)的索引文件。

在邏輯上，可以把一個(gè) Partition 當(dāng)作一個(gè)非常長(zhǎng)的數(shù)組，可通過(guò)這個(gè)“數(shù)組”的索引(Offset)去訪問(wèn)其數(shù)據(jù)。

一方面，由于不同 Partition 可位于不同機(jī)器，因此可以充分利用集群優(yōu)勢(shì)，實(shí)現(xiàn)機(jī)器間的并行處理。

另一方面，由于 Partition 在物理上對(duì)應(yīng)一個(gè)文件夾，即使多個(gè) Partition 位于同一個(gè)節(jié)點(diǎn)，也可通過(guò)配置讓同一節(jié)點(diǎn)上的不同 Partition 置于不同的 disk drive 上，從而實(shí)現(xiàn)磁盤間的并行處理，充分發(fā)揮多磁盤的優(yōu)勢(shì)。

利用多磁盤的具體方法是，將不同磁盤 mount 到不同目錄，然后在 server.properties 中，將 log.dirs 設(shè)置為多目錄(用逗號(hào)分隔)。

Kafka 會(huì)自動(dòng)將所有 Partition 盡可能均勻分配到不同目錄也即不同目錄(也即不同 disk)上。

Partition 是最小并發(fā)粒度，Partition 個(gè)數(shù)決定了可能的并行度。

②ISR 實(shí)現(xiàn)可用性與數(shù)據(jù)一致性的動(dòng)態(tài)平衡

常用數(shù)據(jù)復(fù)制及一致性方案有如下幾種：

Master-Slave：

• RDBMS 的讀寫分離即為典型的 Master-Slave 方案。
• 同步復(fù)制可保證強(qiáng)一致性但會(huì)影響可用性。
• 異步復(fù)制可提供高可用性但會(huì)降低一致性。

WNR：

• 主要用于去中心化的分布式系統(tǒng)中。
• N 代表總副本數(shù)，W 代表每次寫操作要保證的最少寫成功的副本數(shù)，R 代表每次讀至少要讀取的副本數(shù)。
• 當(dāng) W+R>N 時(shí)，可保證每次讀取的數(shù)據(jù)至少有一個(gè)副本擁有新的數(shù)據(jù)。
• 多個(gè)寫操作的順序難以保證，可能導(dǎo)致多副本間的寫操作順序不一致。Dynamo 通過(guò)向量時(shí)鐘保證最終一致性。

Paxos 及其變種：

• Google 的 Chubby，Zookeeper 的原子廣播協(xié)議(Zab)，RAFT 等。

基于 ISR 的數(shù)據(jù)復(fù)制方案：Kafka 的數(shù)據(jù)復(fù)制是以 Partition 為單位的。而多個(gè)備份間的數(shù)據(jù)復(fù)制，通過(guò) Follower 向 Leader 拉取數(shù)據(jù)完成。

從這一點(diǎn)來(lái)講，Kafka 的數(shù)據(jù)復(fù)制方案接近于上文所講的 Master-Slave 方案。

不同的是，Kafka 既不是完全的同步復(fù)制，也不是完全的異步復(fù)制，而是基于 ISR 的動(dòng)態(tài)復(fù)制方案。

ISR，也即 In-Sync Replica。每個(gè) Partition 的 Leader 都會(huì)維護(hù)這樣一個(gè)列表，該列表中，包含了所有與之同步的 Replica(包含 Leader 自己)。

每次數(shù)據(jù)寫入時(shí)，只有 ISR 中的所有 Replica 都復(fù)制完，Leader 才會(huì)將其置為 Commit，它才能被 Consumer 所消費(fèi)。

這種方案，與同步復(fù)制非常接近。但不同的是，這個(gè) ISR 是由 Leader 動(dòng)態(tài)維護(hù)的。

如果 Follower 不能緊“跟上”Leader，它將被 Leader 從 ISR 中移除，待它又重新“跟上”Leader 后，會(huì)被 Leader 再次加到 ISR 中。每次改變 ISR 后，Leader 都會(huì)將新的 ISR 持久化到 Zookeeper 中。

由于 Leader 可移除不能及時(shí)與之同步的 Follower，故與同步復(fù)制相比可避免最慢的 Follower 拖慢整體速度，也即 ISR 提高了系統(tǒng)可用性。

ISR 中的所有 Follower 都包含了所有 Commit 過(guò)的消息，而只有 Commit 過(guò)的消息才會(huì)被 Consumer 消費(fèi)。

故從 Consumer 的角度而言，ISR 中的所有 Replica 都始終處于同步狀態(tài)，從而與異步復(fù)制方案相比提高了數(shù)據(jù)一致性。

ISR 可動(dòng)態(tài)調(diào)整，極限情況下，可以只包含 Leader，極大提高了可容忍的宕機(jī)的 Follower 的數(shù)量。

與 Majority Quorum 方案相比，容忍相同個(gè)數(shù)的節(jié)點(diǎn)失敗，所要求的總節(jié)點(diǎn)數(shù)少了近一半。

③具體實(shí)現(xiàn)層面高效使用磁盤特性和操作系統(tǒng)特性

將寫磁盤的過(guò)程變?yōu)轫樞驅(qū)?/strong>

Kafka 的整個(gè)設(shè)計(jì)中，Partition 相當(dāng)于一個(gè)非常長(zhǎng)的數(shù)組，而 Broker 接收到的所有消息順序?qū)懭脒@個(gè)大數(shù)組中。

同時(shí) Consumer 通過(guò) Offset 順序消費(fèi)這些數(shù)據(jù)，并且不刪除已經(jīng)消費(fèi)的數(shù)據(jù)，從而避免了隨機(jī)寫磁盤的過(guò)程。

由于磁盤有限，不可能保存所有數(shù)據(jù)，實(shí)際上作為消息系統(tǒng) Kafka 也沒(méi)必要保存所有數(shù)據(jù)，需要?jiǎng)h除舊的數(shù)據(jù)。

而這個(gè)刪除過(guò)程，并非通過(guò)使用“讀-寫”模式去修改文件，而是將 Partition 分為多個(gè) Segment，每個(gè) Segment 對(duì)應(yīng)一個(gè)物理文件，通過(guò)刪除整個(gè)文件的方式去刪除 Partition 內(nèi)的數(shù)據(jù)。

這種方式清除舊數(shù)據(jù)的方式，也避免了對(duì)文件的隨機(jī)寫操作。在存儲(chǔ)機(jī)制上，使用了 Log Structured Merge Trees(LSM) 。

注：Log Structured Merge Trees(LSM)，谷歌 “BigTable” 的論文中提出，LSM 是當(dāng)前被用在許多產(chǎn)品的文件結(jié)構(gòu)策略：HBase，Cassandra，LevelDB，SQLite，Kafka。

LSM 被設(shè)計(jì)來(lái)提供比傳統(tǒng)的 B+ 樹或者 ISAM 更好的寫操作吞吐量，通過(guò)消去隨機(jī)的本地更新操作來(lái)達(dá)到這個(gè)目標(biāo)。

這個(gè)問(wèn)題的本質(zhì)還是磁盤隨機(jī)操作慢，順序讀寫快。這兩種操作存在巨大的差距，無(wú)論是磁盤還是 SSD，而且快至少三個(gè)數(shù)量級(jí)。

充分利用 Page Cache

使用 Page Cache 的好處如下：

 
 
 
 

  
  
  
  
buffer = File.readSocket.send(buffer)