為什么要用Message Queue
解耦
在項(xiàng)目啟動(dòng)之初來預(yù)測將來項(xiàng)目會(huì)碰到什么需求孝冒，是極其困難的。消息隊(duì)列在處理過程中間插入了一個(gè)隱含的、基于數(shù)據(jù)的接口層，兩邊的處理過程都要實(shí)現(xiàn)這一接口叫挟。這允許你獨(dú)立的擴(kuò)展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束
冗余
有時(shí)在處理數(shù)據(jù)的時(shí)候處理過程會(huì)失敗限煞。除非數(shù)據(jù)被持久化抹恳，否則將永遠(yuǎn)丟失。消息隊(duì)列把數(shù)據(jù)進(jìn)行持久化直到它們已經(jīng)被完全處理署驻，通過這一方式規(guī)避了數(shù)據(jù)丟失風(fēng)險(xiǎn)奋献。在被許多消息隊(duì)列所采用的”插入-獲取-刪除”范式中，在把一個(gè)消息從隊(duì)列中刪除之前硕舆，需要你的處理過程明確的指出該消息已經(jīng)被處理完畢秽荞，確保你的數(shù)據(jù)被安全的保存直到你使用完畢骤公。
擴(kuò)展性
因?yàn)橄㈥?duì)列解耦了你的處理過程抚官，所以增大消息入隊(duì)和處理的頻率是很容易的；只要另外增加處理過程即可阶捆。不需要改變代碼凌节、不需要調(diào)節(jié)參數(shù)。擴(kuò)展就像調(diào)大電力按鈕一樣簡單洒试。
靈活性 & 峰值處理能力
在訪問量劇增的情況下倍奢，應(yīng)用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量并不常見垒棋；如果為以能處理這類峰值訪問為標(biāo)準(zhǔn)來投入資源隨時(shí)待命無疑是巨大的浪費(fèi)卒煞。使用消息隊(duì)列能夠使關(guān)鍵組件頂住增長的訪問壓力，而不是因?yàn)槌鲐?fù)荷的請求而完全崩潰叼架。
可恢復(fù)性
當(dāng)體系的一部分組件失效畔裕，不會(huì)影響到整個(gè)系統(tǒng)。消息隊(duì)列降低了進(jìn)程間的耦合度乖订，所以即使一個(gè)處理消息的進(jìn)程掛掉扮饶，加入隊(duì)列中的消息仍然可以在系統(tǒng)恢復(fù)后被處理。而這種允許重試或者延后處理請求的能力通常是造就一個(gè)略感不便的用戶和一個(gè)沮喪透頂?shù)挠脩糁g的區(qū)別乍构。
送達(dá)保證
消息隊(duì)列提供的冗余機(jī)制保證了消息能被實(shí)際的處理甜无，只要一個(gè)進(jìn)程讀取了該隊(duì)列即可。在此基礎(chǔ)上，IronMQ提供了一個(gè)”只送達(dá)一次”保證岂丘。無論有多少進(jìn)程在從隊(duì)列中領(lǐng)取數(shù)據(jù)陵究，每一個(gè)消息只能被處理一次。這之所以成為可能奥帘，是因?yàn)楂@取一個(gè)消息只是”預(yù)定”了這個(gè)消息畔乙，暫時(shí)把它移出了隊(duì)列。除非客戶端明確的表示已經(jīng)處理完了這個(gè)消息翩概，否則這個(gè)消息會(huì)被放回隊(duì)列中去牲距，在一段可配置的時(shí)間之后可再次被處理。
順序保證
在許多情況下钥庇，數(shù)據(jù)處理的順序都很重要牍鞠。消息隊(duì)列本來就是排序的，并且能保證數(shù)據(jù)會(huì)按照特定的順序來處理评姨。IronMO保證消息漿糊通過FIFO（先進(jìn)先出）的順序來處理难述，因此消息在隊(duì)列中的位置就是從隊(duì)列中檢索他們的位置。
緩沖
在任何重要的系統(tǒng)中吐句，都會(huì)有需要不同的處理時(shí)間的元素胁后。例如,加載一張圖片比應(yīng)用過濾器花費(fèi)更少的時(shí)間。消息隊(duì)列通過一個(gè)緩沖層來幫助任務(wù)最高效率的執(zhí)行—寫入隊(duì)列的處理會(huì)盡可能的快速嗦枢，而不受從隊(duì)列讀的預(yù)備處理的約束攀芯。該緩沖有助于控制和優(yōu)化數(shù)據(jù)流經(jīng)過系統(tǒng)的速度。
理解數(shù)據(jù)流
在一個(gè)分布式系統(tǒng)里文虏，要得到一個(gè)關(guān)于用戶操作會(huì)用多長時(shí)間及其原因的總體印象侣诺，是個(gè)巨大的挑戰(zhàn)。消息系列通過消息被處理的頻率氧秘，來方便的輔助確定那些表現(xiàn)不佳的處理過程或領(lǐng)域年鸳，這些地方的數(shù)據(jù)流都不夠優(yōu)化。
異步通信
很多時(shí)候丸相，你不想也不需要立即處理消息搔确。消息隊(duì)列提供了異步處理機(jī)制，允許你把一個(gè)消息放入隊(duì)列灭忠，但并不立即處理它膳算。你想向隊(duì)列中放入多少消息就放多少，然后在你樂意的時(shí)候再去處理它們更舞。
常用Message Queue對比
RabbitMQ
RabbitMQ是使用Erlang編寫的一個(gè)開源的消息隊(duì)列畦幢，本身支持很多的協(xié)議：AMQP，XMPP, SMTP, STOMP缆蝉，也正因如此宇葱，它非常重量級瘦真，更適合于企業(yè)級的開發(fā)。同時(shí)實(shí)現(xiàn)了Broker構(gòu)架黍瞧，這意味著消息在發(fā)送給客戶端時(shí)先在中心隊(duì)列排隊(duì)诸尽。對路由，負(fù)載均衡或者數(shù)據(jù)持久化都有很好的支持印颤。
Redis
Redis是一個(gè)基于Key-Value對的NoSQL數(shù)據(jù)庫您机，開發(fā)維護(hù)很活躍。雖然它是一個(gè)Key-Value數(shù)據(jù)庫存儲(chǔ)系統(tǒng)年局，但它本身支持MQ功能际看，所以完全可以當(dāng)做一個(gè)輕量級的隊(duì)列服務(wù)來使用。對于RabbitMQ和Redis的入隊(duì)和出隊(duì)操作矢否，各執(zhí)行100萬次仲闽，每10萬次記錄一次執(zhí)行時(shí)間。測試數(shù)據(jù)分為128Bytes僵朗、512Bytes赖欣、1K和10K四個(gè)不同大小的數(shù)據(jù)。實(shí)驗(yàn)表明：入隊(duì)時(shí)验庙，當(dāng)數(shù)據(jù)比較小時(shí)Redis的性能要高于RabbitMQ顶吮，而如果數(shù)據(jù)大小超過了10K，Redis則慢的無法忍受粪薛；出隊(duì)時(shí)悴了，無論數(shù)據(jù)大小，Redis都表現(xiàn)出非常好的性能汗菜，而RabbitMQ的出隊(duì)性能則遠(yuǎn)低于Redis让禀。
ZeroMQ
ZeroMQ號稱最快的消息隊(duì)列系統(tǒng)，尤其針對大吞吐量的需求場景陨界。ZMQ能夠?qū)崿F(xiàn)RabbitMQ不擅長的高級/復(fù)雜的隊(duì)列，但是開發(fā)人員需要自己組合多種技術(shù)框架痛阻，技術(shù)上的復(fù)雜度是對這MQ能夠應(yīng)用成功的挑戰(zhàn)菌瘪。ZeroMQ具有一個(gè)獨(dú)特的非中間件的模式，你不需要安裝和運(yùn)行一個(gè)消息服務(wù)器或中間件阱当，因?yàn)槟愕膽?yīng)用程序?qū)缪萘诉@個(gè)服務(wù)角色俏扩。你只需要簡單的引用ZeroMQ程序庫，可以使用NuGet安裝弊添，然后你就可以愉快的在應(yīng)用程序之間發(fā)送消息了录淡。但是ZeroMQ僅提供非持久性的隊(duì)列，也就是說如果down機(jī)油坝，數(shù)據(jù)將會(huì)丟失嫉戚。其中刨裆，Twitter的Storm中默認(rèn)使用ZeroMQ作為數(shù)據(jù)流的傳輸。
ActiveMQ
ActiveMQ是Apache下的一個(gè)子項(xiàng)目彬檀。 類似于ZeroMQ帆啃，它能夠以代理人和點(diǎn)對點(diǎn)的技術(shù)實(shí)現(xiàn)隊(duì)列。同時(shí)類似于RabbitMQ窍帝，它少量代碼就可以高效地實(shí)現(xiàn)高級應(yīng)用場景努潘。
Kafka/Jafka
Kafka是Apache下的一個(gè)子項(xiàng)目，是一個(gè)高性能跨語言分布式Publish/Subscribe消息隊(duì)列系統(tǒng)坤学，而Jafka是在Kafka之上孵化而來的疯坤，即Kafka的一個(gè)升級版。具有以下特性：快速持久化深浮，可以在O(1)的系統(tǒng)開銷下進(jìn)行消息持久化贴膘；高吞吐，在一臺普通的服務(wù)器上既可以達(dá)到10W/s的吞吐速率略号；完全的分布式系統(tǒng)刑峡，Broker、Producer玄柠、Consumer都原生自動(dòng)支持分布式突梦，自動(dòng)實(shí)現(xiàn)負(fù)載均衡；支持Hadoop數(shù)據(jù)并行加載羽利，對于像Hadoop的一樣的日志數(shù)據(jù)和離線分析系統(tǒng)宫患，但又要求實(shí)時(shí)處理的限制，這是一個(gè)可行的解決方案这弧。Kafka通過Hadoop的并行加載機(jī)制來統(tǒng)一了在線和離線的消息處理娃闲，這一點(diǎn)也是本課題所研究系統(tǒng)所看重的。Apache Kafka相對于ActiveMQ是一個(gè)非常輕量級的消息系統(tǒng)匾浪，除了性能非常好之外皇帮，還是一個(gè)工作良好的分布式系統(tǒng)。
RocketMq
它是阿里的一個(gè)開源項(xiàng)目蛋辈，大廠加持属拾，他的質(zhì)量是有保證的。他采取推冷溶、拉的模式渐白，支持事務(wù)消息，集成了Spring cloud逞频。
kafka組件概念
Broker
Kafka集群包含一個(gè)或多個(gè)服務(wù)器纯衍，這種服務(wù)器被稱為broker
Topic
每條發(fā)布到Kafka集群的消息都有一個(gè)類別，這個(gè)類別被稱為topic苗胀。（物理上不同topic的消息分開存儲(chǔ)襟诸，邏輯上一個(gè)topic的消息雖然保存于一個(gè)或多個(gè)broker上但用戶只需指定消息的topic即可生產(chǎn)或消費(fèi)數(shù)據(jù)而不必關(guān)心數(shù)據(jù)存于何處）
Partition
parition是物理上的概念瓦堵，每個(gè)topic包含一個(gè)或多個(gè)partition，創(chuàng)建topic時(shí)可指定parition數(shù)量励堡。每個(gè)partition對應(yīng)于一個(gè)文件夾谷丸，該文件夾下存儲(chǔ)該partition的數(shù)據(jù)和索引文件
Producer
負(fù)責(zé)發(fā)布消息到Kafka broker
Consumer
消費(fèi)消息。每個(gè)consumer屬于一個(gè)特定的consuer group（可為每個(gè)consumer指定group name应结，若不指定group name則屬于默認(rèn)的group）刨疼。使用consumer high level API時(shí)，同一topic的一條消息只能被同一個(gè)consumer group內(nèi)的一個(gè)consumer消費(fèi)鹅龄，但多個(gè)consumer group可同時(shí)消費(fèi)這一消息揩慕。
Push vs. Pull
作為一個(gè)messaging system，Kafka遵循了傳統(tǒng)的方式扮休，選擇由producer向broker push消息并由consumer從broker pull消息迎卤。一些logging-centric system，比如Facebook的 Scribe 和Cloudera的 Flume ,采用非常不同的push模式玷坠。事實(shí)上蜗搔，push模式和pull模式各有優(yōu)劣。
push模式很難適應(yīng)消費(fèi)速率不同的消費(fèi)者八堡，因?yàn)橄l(fā)送速率是由broker決定的樟凄。push模式的目標(biāo)是盡可能以最快速度傳遞消息，但是這樣很容易造成consumer來不及處理消息兄渺，典型的表現(xiàn)就是拒絕服務(wù)以及網(wǎng)絡(luò)擁塞缝龄。而pull模式則可以根據(jù)consumer的消費(fèi)能力以適當(dāng)?shù)乃俾氏M(fèi)消息。
Consumer Rebalance
（本節(jié)所講述內(nèi)容均基于Kafka consumer high level API）
Kafka保證同一consumer group中只有一個(gè)consumer會(huì)消息某條消息挂谍，實(shí)際上叔壤，Kafka保證的是穩(wěn)定狀態(tài)下每一個(gè)consumer實(shí)例只會(huì)消費(fèi)某一個(gè)或多個(gè)特定partition的數(shù)據(jù)，而某個(gè)partition的數(shù)據(jù)只會(huì)被某一個(gè)特定的consumer實(shí)例所消費(fèi)口叙。這樣設(shè)計(jì)的劣勢是無法讓同一個(gè)consumer group里的consumer均勻消費(fèi)數(shù)據(jù)炼绘，優(yōu)勢是每個(gè)consumer不用都跟大量的broker通信，減少通信開銷庐扫，同時(shí)也降低了分配難度饭望，實(shí)現(xiàn)也更簡單。另外形庭，因?yàn)橥粋€(gè)partition里的數(shù)據(jù)是有序的，這種設(shè)計(jì)可以保證每個(gè)partition里的數(shù)據(jù)也是有序被消費(fèi)厌漂。
如果某consumer group中consumer數(shù)量少于partition數(shù)量萨醒，則至少有一個(gè)consumer會(huì)消費(fèi)多個(gè)partition的數(shù)據(jù)，如果consumer的數(shù)量與partition數(shù)量相同苇倡，則正好一個(gè)consumer消費(fèi)一個(gè)partition的數(shù)據(jù)富纸，而如果consumer的數(shù)量多于partition的數(shù)量時(shí)囤踩，會(huì)有部分consumer無法消費(fèi)該topic下任何一條消息。
如下例所示晓褪，如果topic1有0堵漱，1，2共三個(gè)partition涣仿，當(dāng)group1只有一個(gè)consumer(名為consumer1)時(shí)勤庐，該 consumer可消費(fèi)這3個(gè)partition的所有數(shù)據(jù)。
增加一個(gè)consumer(consumer2)后好港，其中一個(gè)consumer（consumer1）可消費(fèi)2個(gè)partition的數(shù)據(jù)愉镰，另外一個(gè)consumer(consumer2)可消費(fèi)另外一個(gè)partition的數(shù)據(jù)。
再增加一個(gè)consumer(consumer3)后钧汹，每個(gè)consumer可消費(fèi)一個(gè)partition的數(shù)據(jù)丈探。consumer1消費(fèi)partition0，consumer2消費(fèi)partition1拔莱，consumer3消費(fèi)partition2
再增加一個(gè)consumer（consumer4）后碗降，其中3個(gè)consumer可分別消費(fèi)一個(gè)partition的數(shù)據(jù)，另外一個(gè)consumer（consumer4）不能消費(fèi)topic1任何數(shù)據(jù)塘秦。
此時(shí)關(guān)閉consumer1讼渊，剩下的consumer可分別消費(fèi)一個(gè)partition的數(shù)據(jù)。
接著關(guān)閉consumer2嗤形，剩下的consumer3可消費(fèi)2個(gè)partition精偿，consumer4可消費(fèi)1個(gè)partition。
再關(guān)閉consumer3赋兵，剩下的consumer4可同時(shí)消費(fèi)topic1的3個(gè)partition笔咽。
[圖片上傳中...(image-2e51af-1595025465335-0)]
consumer rebalance算法如下：
- Sort PT (all partitions in topic T)
- Sort CG(all consumers in consumer group G)
- Let i be the index position of Ci in CG and let N=size(PT)/size(CG)
- Remove current entries owned by Ci from the partition owner registry
- Assign partitions from i N to (i+1) N-1 to consumer Ci
- Add newly assigned partitions to the partition owner registry
消息Deliver guarantee
通過上文介紹，想必讀者已經(jīng)明天了producer和consumer是如何工作的霹期，以及Kafka是如何做replication的叶组，接下來要討論的是Kafka如何確保消息在producer和consumer之間傳輸。有這么幾種可能的delivery guarantee：
At most once 消息可能會(huì)丟历造，但絕不會(huì)重復(fù)傳輸
At least one 消息絕不會(huì)丟甩十，但可能會(huì)重復(fù)傳輸
Exactly once 每條消息肯定會(huì)被傳輸一次且僅傳輸一次，很多時(shí)候這是用戶所想要的吭产。
Kafka的delivery guarantee semantic非常直接侣监。當(dāng)producer向broker發(fā)送消息時(shí)，一旦這條消息被commit臣淤，因數(shù)replication的存在橄霉，它就不會(huì)丟。但是如果producer發(fā)送數(shù)據(jù)給broker后邑蒋，遇到的網(wǎng)絡(luò)問題而造成通信中斷姓蜂，那producer就無法判斷該條消息是否已經(jīng)commit按厘。這一點(diǎn)有點(diǎn)像向一個(gè)自動(dòng)生成primary key的數(shù)據(jù)庫表中插入數(shù)據(jù)。雖然Kafka無法確定網(wǎng)絡(luò)故障期間發(fā)生了什么钱慢，但是producer可以生成一種類似于primary key的東西逮京，發(fā)生故障時(shí)冪等性的retry多次，這樣就做到了Exactly one 束莫。截止到目前(Kafka 0.8.2版本懒棉，2015-01-25)，這一feature還并未實(shí)現(xiàn)麦箍，有希望在Kafka未來的版本中實(shí)現(xiàn)漓藕。（所以目前默認(rèn)情況下一條消息從producer和broker是確保了 At least once ，但可通過設(shè)置producer異步發(fā)送實(shí)現(xiàn) At most once ）挟裂。
接下來討論的是消息從broker到consumer的delivery guarantee semantic享钞。（僅針對Kafka consumer high level API）。consumer在從broker讀取消息后诀蓉，可以選擇commit栗竖，該操作會(huì)在Zookeeper中存下該consumer在該partition下讀取的消息的offset。該consumer下一次再讀該partition時(shí)會(huì)從下一條開始讀取渠啤。如未commit狐肢，下一次讀取的開始位置會(huì)跟上一次commit之后的開始位置相同。當(dāng)然可以將consumer設(shè)置為autocommit沥曹，即consumer一旦讀到數(shù)據(jù)立即自動(dòng)commit份名。如果只討論這一讀取消息的過程，那Kafka是確保了 Exactly once 妓美。但實(shí)際上實(shí)際使用中consumer并非讀取完數(shù)據(jù)就結(jié)束了僵腺，而是要進(jìn)行進(jìn)一步處理，而數(shù)據(jù)處理與commit的順序在很大程度上決定了消息從broker和consumer的delivery guarantee semantic壶栋。
讀完消息先commit再處理消息辰如。這種模式下，如果consumer在commit后還沒來得及處理消息就crash了贵试，下次重新開始工作后就無法讀到剛剛已提交而未處理的消息琉兜，這就對應(yīng)于 At most once
讀完消息先處理再commit。這種模式下毙玻，如果處理完了消息在commit之前consumer crash了豌蟋，下次重新開始工作時(shí)還會(huì)處理剛剛未commit的消息，實(shí)際上該消息已經(jīng)被處理過了桑滩。這就對應(yīng)于 At least once 夺饲。在很多情況使用場景下，消息都有一個(gè)primary key施符，所以消息的處理往往具有冪等性往声，即多次處理這一條消息跟只處理一次是等效的，那就可以認(rèn)為是 Exactly once 戳吝。（人個(gè)感覺這種說法有些牽強(qiáng)浩销，畢竟它不是Kafka本身提供的機(jī)制，而且primary key本身不保證操作的冪等性听哭。而且實(shí)際上我們說delivery guarantee semantic是討論被處理多少次慢洋，而非處理結(jié)果怎樣，因?yàn)樘幚矸绞蕉喾N多樣陆盘，我們的系統(tǒng)不應(yīng)該把處理過程的特性—如是否冪等性普筹，當(dāng)成Kafka本身的feature）
如果一定要做到 Exactly once ，就需要協(xié)調(diào)offset和實(shí)際操作的輸出隘马。精典的做法是引入兩階段提交太防。如果能讓offset和操作輸入存在同一個(gè)地方，會(huì)更簡潔和通用酸员。這種方式可能更好蜒车，因?yàn)樵S多輸出系統(tǒng)可能不支持兩階段提交。比如幔嗦，consumer拿到數(shù)據(jù)后可能把數(shù)據(jù)放到HDFS酿愧，如果把最新的offset和數(shù)據(jù)本身一起寫到HDFS，那就可以保證數(shù)據(jù)的輸出和offset的更新要么都完成邀泉，要么都不完成嬉挡，間接實(shí)現(xiàn) Exactly once 。（目前就high level API而言汇恤，offset是存于Zookeeper中的庞钢，無法存于HDFS，而low level API的offset是由自己去維護(hù)的屁置，可以將之存于HDFS中）
總之焊夸，Kafka默認(rèn)保證 At least once ，并且允許通過設(shè)置producer異步提交來實(shí)現(xiàn) At most once 蓝角。而 Exactly once 要求與目標(biāo)存儲(chǔ)系統(tǒng)協(xié)作阱穗，幸運(yùn)的是Kafka提供的offset可以使用這種方式非常直接非常容易。
