apache kafka參考

http://kafka.apache.org/documentation.html

消息隊列方式：

點對點：

消息生產者生產消息發送到queue中，然后消息消費者從queue中取出并且消費消息。這里要注意：

• 消息被消費以后，queue中不再有存儲，所以消息消費者不可能消費到已經被消費的消息。
• Queue支持存在多個消費者，但是對一個消息而言，只會有一個消費者可以消費。

發布/訂閱：

消息生產者（發布）將消息發布到topic中，同時有多個消息消費者（訂閱）消費該消息。和點對點方式不同，發布到topic的消息會被所有訂閱者消費。

背景介紹：

Kafka 是一個消息系統，原本開發自 LinkedIn，用作 LinkedIn 的活動流（Activity Stream）和運營數據處理管道（Pipeline）的基礎。現在它已被多家公司作為多種類型的數據管道和消息系統使用。

活動流數據是幾乎所有站點在對其網站使用情況做報表時都要用到的數據中最常規的部分。活動數據包括頁面訪問量（Page View）、被查看內容方面的信息以及搜索情況等內容。這種數據通常的處理方式

是先把各種活動以日志的形式寫入某種文件，然后周期性地對這些文件進行統計分析。運營數據指的是服務器的性能數據（CPU、IO 使用率、請求時間、服務日志等等數據)，總的來說，運營數據的統計方法種類繁多。

如上圖所示，一個典型的Kafka集群中包含若干Producer（可以是web前端產生的Page View，或者是服務器日志，系統CPU、Memory等），若干broker（Kafka支持水平擴展，一般broker數量越多，

集群吞吐率越高），若干Consumer Group，以及一個Zookeeper集群。Kafka通過Zookeeper管理集群配置，選舉leader，以及在Consumer Group發生變化時進行rebalance。Producer使用push模式

將消息發布到broker，Consumer使用pull模式從broker訂閱并消費消息。

kafka名詞解釋和工作方式：

• Broker ： 一臺kafka服務器就是一個broker。一個集群由多個broker組成。一個broker可以容納多個topic。
• Topic ： 一類消息，例如page view日志、click日志等都可以以topic的形式存在。咱們可以理解為一個隊列。
• Producer ： 消息生產者，就是向kafka broker推送消息的客戶端。
• Consumer ： 消息消費者，向kafka broker拉取消息的客戶端
• Consumer Group （CG）：這是kafka用來實現一個topic消息的廣播（發給所有的consumer）和單播（發給任意一個consumer）的手段。一個topic可以有多個CG。

??????????????????????????????????????????????? topic的消息會復制（不是真的復制，是概念上的）到所有的CG，但每個CG只會把消息發給該CG中的一個consumer。

??????????????????????????????????????????????? 如果需要實現廣播，只要每個consumer有一個獨立的CG就可以了。

??????????????????????????????????????????????? 要實現單播只要所有的consumer在同一個CG。

??????????????????????????????????????????????? 用CG還可以將consumer進行自由的分組而不需要多次發送消息到不同的topic。

• Partition：為了實現擴展性，一個非常大的topic可以分布到多個broker（即服務器）上，一個topic可以分為多個partition，每個partition是一個有序的隊列。partition中的每條消息都會被分配一個有序的id（offset）。

?????????????????????? kafka只保證按一個partition中的順序將消息發給consumer，不保證一個topic的整體（多個partition間）的順序。

• Offset：每個partition都由一系列有序的、不可變的消息組成，這些消息被連續的追加到partition中。partition中的每個消息都有一個連續的序列號叫做offset,用于partition唯一標識一條消息。

?????????????????? kafka的存儲文件都是按照offset.kafka來命名，用offset做名字的好處是方便查找。例如你想找位于2049的位置，只要找到2048.kafka的文件即可。當然the first offset就是00000000000.kafka

kafka特性：

• 通過O(1)的磁盤數據結構提供消息的持久化，這種結構對于即使數以TB的消息存儲也能夠保持長時間的穩定性能。
• 高吞吐量：即使是非常普通的硬件kafka也可以支持每秒數十萬的消息。
• 支持同步和異步復制兩種HA
• Consumer客戶端pull，隨機讀,利用sendfile系統調用，zero-copy ,批量拉數據
• 消費狀態保存在客戶端
• 消息存儲順序寫
• 數據遷移、擴容對用戶透明
• 支持Hadoop并行數據加載。
• 支持online和offline的場景。
• 持久化：通過將數據持久化到硬盤以及replication防止數據丟失。
• scale out：無需停機即可擴展機器。
• 定期刪除機制，支持設定partitions的segment file保留時間。

可靠性（一致性)

kafka(MQ)要實現從producer到consumer之間的可靠的消息傳送和分發。傳統的MQ系統通常都是通過broker和consumer間的確認（ack）機制實現的，并在broker保存消息分發的狀態。

即使這樣一致性也是很難保證的（參考原文）。www.leyouzaixan.cn kafka的做法是由consumer自己保存狀態，也不要任何確認。這樣雖然consumer負擔更重，但其實更靈活了。

因為不管consumer上任何原因導致需要重新處理消息，都可以再次從broker獲得。

kafak系統擴展性

kafka使用zookeeper來實現動態的集群擴展，不需要更改客戶端（producer和consumer）的配置。broker會在zookeeper注冊并保持相關的元數據（topic，partition信息等）更新。

而客戶端會在zookeeper上注冊相關的watcher。一旦zookeeper發生變化，客戶端能及時感知并作出相應調整。這樣就保證了添加或去除broker時，各broker間仍能自動實現負載均衡。

kafka設計目標

高吞吐量是其核心設計之一。

• 數據磁盤持久化：消息不在內存中cache，直接寫入到磁盤，充分利用磁盤的順序讀寫性能。
• zero-copy：減少IO操作步驟。
• 支持數據批量發送和拉取。
• 支持數據壓縮。
• Topic劃分為多個partition，提高并行處理能力。

Producer負載均衡和HA機制

• producer根據用戶指定的算法，將消息發送到指定的partition。
• 存在多個partiiton，每個partition有自己的replica，每個replica分布在不同的Broker節點上。
• 多個partition需要選取出lead partition，www.uuweb.cn lead www.yigouyule2.cn ? ?partition負責讀寫，并由zookeeper負責fail over。
• 通過zookeeper管理broker與consumer的動態加入與離開。

Consumer的pull機制

由于kafka broker會持久化數據，broker沒有cache壓力，因此，consumer比較適合采取pull的方式消費數據，具體特別如下：

• 簡化kafka設計，降低了難度。
• Consumer根據消費能力自主控制消息拉取速度。
• consumer根據自身情況自主選擇消費模式，例如批量，重復消費，從制定partition或位置(offset)開始消費等.

Consumer與topic關系以及機制

本質上kafka只支持Topic.每個consumer屬于一個consumer group;反過來說,每個group中可以有多個consumer.對于Topic中的一條特定的消息,?
只會被訂閱此Topic的每個group中的一個consumer消費,此消息不會發送給一個group的多個consumer;那么一個group中所有的consumer將會交錯的消費整個Topic.?
如果所有的consumer都具有相同的group,這種情況和JMS(Java Message Service) queue模式很像;消息將會在consumers之間負載均衡.?
如果所有的consumer都具有不同的group,那這就是"發布-訂閱";消息將會廣播給所有的消費者.

在kafka中,一個partition中的消息只會被www.taohuayuangw.com group中的一個consumer消費(同一時刻);每個group中consumer消息消費互相獨立;我們可以認為一個group是一個"訂閱"者,

一個Topic中的每個partions,只會被一個"訂閱者"中的一個consumer消費,不過一個consumer可以同時消費多個partitions中的消息.

kafka只能保證一個partition中的消息被某個consumer消費時是順序的.事實上,從Topic角度來說,當有多個partitions時,消息仍不是全局有序的.

通常情況下,一個group中會包含多個consumer,這樣不僅可以提高topic中消息的并發消費能力,而且還能提高"故障容錯"性,如果group中的某個consumer失效,

那么其消費的partitions將會有其他consumer自動接管.kafka的設計原理決定,對于一個topic,同一個group中不能有多于partitions個數的consumer同時消費,

否則將意味著某些consumer將無法得到消息.

Producer均衡算法

kafka集群中的任何一個broker,都可以向producer提供metadata信息,這些metadata中包含"集群中存活的servers列表"/"partitions leader列表"?
等信息(請參看zookeeper中的節點信息).當producer獲取到metadata信心之后, producer將會和Topic下所有partition leader保持socket連接;?
消息由producer直接通過socket發送到broker,中間不會經過任何"路由層".事實上,消息被路由到哪個partition上,有producer客戶端決定.?
比如可以采用"random""key-hash""輪詢"等,如果一個topic中有多個partitions,那么在producer端實現"消息均衡分發"是必要的.?
在producer端的配置文件中,開發者可以指定partition路由的方式.

Consumer均衡算法

當一個group中,有consumer加入或者離開時,會觸發partitions均衡.均衡的最終目的,是提升topic的并發消費能力.?
1) 假如topic1,具有如下partitions: P0,P1,P2,P3?
2) 加入group中,有如下consumer: C0,C1?
3) 首先根據partition索引號對partitions排序: P0,P1,P2,P3?
4) 根據consumer.id排序: C0,C1?
5) 計算倍數: M = [P0,P1,P2,P3].size / [C0,C1].size,本例值M=2(向上取整)?
6) 然后依次分配partitions: C0 = [P0,P1],C1=[P2,P3],即Ci = [P(i * M),P((i + 1) * M -1)]

kafka broker集群內broker之間replication機制

kafka中,replication策略是基于partition,而不是topic;kafka將每個partition數據復制到多個server上,任何一個partition有一個leader和多個follower(可以沒有);

備份的個數可以通過broker配置文件來設定.leader處理所有的read-write請求,follower需要和leader保持同步.Follower就像一個"consumer",

消費消息并保存在本地日志中;leader負責跟蹤所有的follower狀態,如果follower"落后"太多或者失效,leader將會把它從replicas同步列表中刪除.

當所有的follower都將一條消息保存成功,此消息才被認為是"committed",那么此時consumer才能消費它,這種同步策略,就要求follower和leader之間必須具有良好的網絡環境.

即使只有一個replicas實例存活,仍然可以保證消息的正常發送和接收,只要zookeeper集群存活即可.(備注:不同于其他分布式存儲,比如hbase需要"多數派"存活才行)

kafka判定一個follower存活與否的條件有2個:

1) follower需要和zookeeper保持良好的鏈接???

2) 它必須能夠及時的跟進leader,不能落后太多.

如果同時滿足上述2個條件,那么leader就認為此follower是"活躍的".如果一個follower失效(server失效)或者落后太多,

leader將會把它從同步列表中移除’備注:如果此replicas落后太多,它將會繼續從leader中fetch數據,直到足夠up-to-date,

然后再次加入到同步列表中;kafka不會更換replicas宿主,因為"同步列表"中replicas需要足夠快,這樣才能保證producer發布消息時接受到ACK的延遲較小。

當leader失效時，需在followers中選取出新的leader,可能此時follower落后于leader,因此需要選擇一個"up-to-date"的follower.kafka中leader選舉并沒有采用"投票多數派"的算法,

因為這種算法對于"網絡穩定性"/"投票參與者數量"等條件有較高的要求,而且kafka集群的設計,還需要容忍N-1個replicas失效.對于kafka而言,

每個partition中所有的replicas信息都可以在zookeeper中獲得,那么選舉leader將是一件非常簡單的事情.選擇follower時需要兼顧一個問題,

就是新leader server上所已經承載的partition leader的個數,如果一個server上有過多的partition leader,意味著此server將承受著更多的IO壓力.

在選舉新leader,需要考慮到"負載均衡",partition leader較少的broker將會更有可能成為新的leader.

在整幾個集群中,只要有一個replicas存活,那么此partition都可以繼續接受讀寫操作.

總結:

??? 1) Producer端直接連接broker.list列表,從列表中返回TopicMetadataResponse,該Metadata包含Topic下每個partition leader建立socket連接并發送消息.

??? 2) Broker端使用zookeeper用來注冊broker信息,以及監控partition leader存活性.

? ?3) Consumer端使用zookeeper用來注冊consumer信息,其中包括consumer消費的partition列表等,同時也用來發現broker列表,并和partition leader建立socket連接,并獲取消息.

總結

以上是生活随笔為你收集整理的kafka 消息服务的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。