培訓(xùn)課程Kafka.ppt

1、深入淺出hadoop,課程安排,Kafka是什么 kafka體系結(jié)構(gòu) kafka設(shè)計(jì)理念簡介* kafka通信協(xié)議 kafka的偽分布安裝、集群安裝* kafka的shell操作、java操作* kafka設(shè)計(jì)理念* kafka producer和consumer開發(fā)*,Kafka產(chǎn)生背景,Kafka 是分布式發(fā)布-訂閱消息系統(tǒng)。它最初由 LinkedIn 公司開發(fā)，使用 Scala語言編寫,之后成為 Apache 項(xiàng)目的一部分。Kafka 是一個(gè)分布式的，可劃分的，多訂閱者,冗余備份的持久性的日志服務(wù)。它主要用于處理活躍的流式數(shù)據(jù)。 在大數(shù)據(jù)系統(tǒng)中，常常會(huì)碰到一個(gè)問題，整個(gè)大數(shù)據(jù)是由各個(gè)子系

2、統(tǒng)組成，數(shù)據(jù)需要在各個(gè)子系統(tǒng)中高性能，低延遲的不停流轉(zhuǎn)。傳統(tǒng)的企業(yè)消息系統(tǒng)并 不是非常適合大規(guī)模的數(shù)據(jù)處理。為了已在同時(shí)搞定在線應(yīng)用（消息）和離線應(yīng)用（數(shù)據(jù)文件，日志）Kafka 就出現(xiàn)了。Kafka 可以起到兩個(gè)作用： 降低系統(tǒng)組網(wǎng)復(fù)雜度 降低編程復(fù)雜度，各個(gè)子系統(tǒng)不在是相互協(xié)商接口，各個(gè)子系統(tǒng)類似插口插在插座上，Kafka 承擔(dān)高速數(shù)據(jù)總線的作用。 kafka系列文章索引:,Kafka簡介,同時(shí)為發(fā)布和訂閱提供高吞吐量。據(jù)了解，Kafka 每秒可以生產(chǎn)約 25 萬消息（50 MB），每秒處理 55 萬消息（110 MB）。 可進(jìn)行持久化操作。將消息持久化到磁盤，因此可用于批量消費(fèi)，例如 E

3、TL，以及實(shí)時(shí)應(yīng)用程序。通過將數(shù)據(jù)持久化到硬盤以及 replication 防止數(shù)據(jù)丟失。 分布式系統(tǒng)，易于向外擴(kuò)展。所有的 producer、broker 和 consumer 都會(huì)有多個(gè)，均為分布式的。無需停機(jī)即可擴(kuò)展機(jī)器。 消息被處理的狀態(tài)是在 consumer 端維護(hù)，而不是由 server 端維護(hù)。當(dāng)失敗時(shí)能自動(dòng)平衡。 支持 online 和 offline 的場景。,Kafka的簡介,設(shè)計(jì)關(guān)注重點(diǎn)： 為生產(chǎn)者和消費(fèi)者提供一個(gè)通用的API 消息的持久化 高吞吐量，可以滿足百萬級(jí)別消息處理 對(duì)分布式和高擴(kuò)展性的支持 kafka最基本的架構(gòu)是生產(chǎn)者發(fā)布一個(gè)消息到Kafka的一個(gè)主題（to

4、pic），這個(gè)主題即是由扮演KafkaServer角色的broker提供，消費(fèi)者訂閱這個(gè)主題，然后從中獲取消息. Kafka是如何解決查找效率的的問題呢？,Kafka的兩大法寶,數(shù)據(jù)文件的分段： Kafka解決查詢效率的手段之一是將數(shù)據(jù)文件分段； 為數(shù)據(jù)文件建索引： 索引優(yōu)化：稀疏存儲(chǔ)，每隔一定字節(jié)的數(shù)據(jù)建立一條索引。,消息隊(duì)列分類,點(diǎn)對(duì)點(diǎn): 消息生產(chǎn)者生產(chǎn)消息發(fā)送到queue中，然后消息消費(fèi)者從queue中取出并且消費(fèi)消息。 注意： 消息被消費(fèi)以后，queue中不再有存儲(chǔ)，所以消息消費(fèi)者不可能消費(fèi)到已經(jīng)被消費(fèi)的消息。 Queue支持存在多個(gè)消費(fèi)者，但是對(duì)一個(gè)消息而言，只會(huì)有一個(gè)消費(fèi)者可以消費(fèi)

5、。 發(fā)布/訂閱: 消息生產(chǎn)者（發(fā)布）將消息發(fā)布到topic中，同時(shí)有多個(gè)消息消費(fèi)者（訂閱）消費(fèi)該消息。和點(diǎn)對(duì)點(diǎn)方式不同，發(fā)布到topic的消息會(huì)被所有訂閱者消費(fèi)。,消息隊(duì)列MQ對(duì)比,RabbitMQ：支持的協(xié)議多，非常重量級(jí)消息隊(duì)列，對(duì)路由(Routing)，負(fù)載均衡(Load balance)或者數(shù)據(jù)持久化都有很好的支持。 ZeroMQ：號(hào)稱最快的消息隊(duì)列系統(tǒng)，尤其針對(duì)大吞吐量的需求場景，擅長的高級(jí)/復(fù)雜的隊(duì)列，但是技術(shù)也復(fù)雜，并且只提供非持久性的隊(duì)列。 ActiveMQ：Apache下的一個(gè)子項(xiàng)，類似ZeroMQ，能夠以代理人和點(diǎn)對(duì)點(diǎn)的技術(shù)實(shí)現(xiàn)隊(duì)列。 Redis：是一個(gè)key-Value的

6、NOSql數(shù)據(jù)庫，但也支持MQ功能，數(shù)據(jù)量較小，性能優(yōu)于RabbitMQ，數(shù)據(jù)超過10K就慢的無法忍受,Kafka部署架構(gòu),Kafka集群架構(gòu),Kafka的基本概念,Topic：特指 Kafka 處理的消息源（feeds of messages）的不同分類。 Partition：Topic 物理上的分組，一個(gè) topic 可以分為多個(gè) partition，每個(gè) partition 是一個(gè)有序的隊(duì)列。partition 中的每條消息都會(huì)被分配一個(gè)有序的 id（offset）。 Message：消息，是通信的基本單位，每個(gè) producer 可以向一個(gè) topic（主題）發(fā)布一些消息。 Produ

7、cers：消息和數(shù)據(jù)生產(chǎn)者，向 Kafka 的一個(gè) topic 發(fā)布消息的過程叫做 producers。 Consumers：消息和數(shù)據(jù)消費(fèi)者，訂閱 topics 并處理其發(fā)布的消息的過程叫做 consumers。 Broker：緩存代理，Kafka 集群中的一臺(tái)或多臺(tái)服務(wù)器統(tǒng)稱為 broker。,Kafka的Producers,Producer將消息發(fā)布到指定的Topic中,同時(shí)Producer也能決定將此消息歸屬于哪個(gè)partition;比如基于round-robin方式或者通過其他的一些算法等. 消息和數(shù)據(jù)生產(chǎn)者，向 Kafka 的一個(gè) topic 發(fā)布消息的過程叫做 producers

8、。 異步發(fā)送 批量發(fā)送可以很有效的提高發(fā)送效率。Kafka producer的異步發(fā)送模式允許進(jìn)行批量發(fā)送，先將消息緩存在內(nèi)存中，然后一次請(qǐng)求批量發(fā)送出去。,Kafka的Broker,Broker：緩存代理，Kafka 集群中的一臺(tái)或多臺(tái)服務(wù)器統(tǒng)稱為 broker。 為了減少磁盤寫入的次數(shù),broker會(huì)將消息暫時(shí)buffer起來,當(dāng)消息的個(gè)數(shù)(或尺寸)達(dá)到一定閥值時(shí),再flush到磁盤,這樣減少了磁盤IO調(diào)用的次數(shù)。,Kafka的broker無狀態(tài)機(jī)制,1. Broker沒有副本機(jī)制，一旦broker宕機(jī)，該broker的消息將都不可用。 2. Broker不保存訂閱者的狀態(tài)，由訂閱者自己保

9、存。 3. 無狀態(tài)導(dǎo)致消息的刪除成為難題（可能刪除的消息正在被訂閱），kafka采用基于時(shí)間的SLA(服務(wù)水平保證)，消息保存一定時(shí)間（通常為7天）后會(huì)被刪除。 4. 消息訂閱者可以rewind back到任意位置重新進(jìn)行消費(fèi)，當(dāng)訂閱者故障時(shí)，可以選擇最小的offset(id)進(jìn)行重新讀取消費(fèi)消息。,Kafka的Consumers,消息和數(shù)據(jù)消費(fèi)者，訂閱 topics 并處理其發(fā)布的消息的過程叫做 consumers。 本質(zhì)上kafka只支持Topic.每個(gè)consumer屬于一個(gè)consumer group;反過來說,每個(gè)group中可以有多個(gè)consumer.發(fā)送到Topic的消息,只會(huì)被

10、訂閱此Topic的每個(gè)group中的一個(gè)consumer消費(fèi). 在 kafka中,我們 可以認(rèn)為一個(gè)group是一個(gè)訂閱者,一個(gè)Topic中的每個(gè)partions,只會(huì)被一個(gè)訂閱者中的一個(gè)consumer消費(fèi),不過一個(gè) consumer可以消費(fèi)多個(gè)partitions中的消息.kafka只能保證一個(gè)partition中的消息被某個(gè)consumer消費(fèi)時(shí),消息是順 序的.事實(shí)上,從Topic角度來說,消息仍不是有序的. 注： kafka的設(shè)計(jì)原理決定,對(duì)于一個(gè)topic,同一個(gè)group中不能有多于partitions個(gè)數(shù)的consumer同時(shí)消費(fèi),否則將意味著某些consumer將無法得到消息

11、.,Kafka的Consumer group,1. 允許consumer group（包含多個(gè)consumer，如一個(gè)集群同時(shí)消費(fèi)）對(duì)一個(gè)topic進(jìn)行消費(fèi)，不同的consumer group之間獨(dú)立訂閱。 2. 為了對(duì)減小一個(gè)consumer group中不同consumer之間的分布式協(xié)調(diào)開銷，指定partition為最小的并行消費(fèi)單位，即一個(gè)group內(nèi)的consumer只能消費(fèi)不同的partition。,Kafka的Topics/Log,一個(gè)Topic可以認(rèn)為是一類消息，每個(gè)topic將被分成多partition(區(qū)),每個(gè)partition在存儲(chǔ)層面是append log文件。任何發(fā)

12、布到此partition的消息都會(huì)被直接追加到log文件的尾部，每條消息在文件中的位置稱為offset（偏移量）,partition是以文件的形式存儲(chǔ)在文件系統(tǒng)中。 Logs文件根據(jù)broker中的配置要求,保留一定時(shí)間后刪除來釋放磁盤空間。,Partition： Topic 物理上的分組，一個(gè) topic 可以分為多個(gè) partition，每個(gè) partition 是一個(gè)有序的隊(duì)列。 partition 中的每條消息都會(huì)被分配一個(gè)有序的 id（offset）。,Kafka的partitions,設(shè)計(jì)目的： kafka基于文件存儲(chǔ).通過分區(qū),可以將日志內(nèi)容分散到多個(gè)server上,來避免文件尺

13、寸達(dá)到單機(jī)磁盤的上限,每個(gè)partiton都會(huì)被當(dāng)前server(kafka實(shí)例)保存; 可以將一個(gè)topic切分多任意多個(gè)partitions,來消息保存/消費(fèi)的效率. 越多的partitions意味著可以容納更多的consumer,有效提升并發(fā)消費(fèi)的能力.,Kafka的Message,Message消息：是通信的基本單位，每個(gè) producer 可以向一個(gè) topic（主題）發(fā)布一些消息。 Kafka中的Message是以topic為基本單位組織的，不同的topic之間是相互獨(dú)立的。每個(gè)topic又可以分成幾個(gè)不同的partition(每個(gè)topic有幾個(gè)partition是在創(chuàng)建topi

14、c時(shí)指定的)，每個(gè)partition存儲(chǔ)一部分Message。 partition中的每條Message包含了以下三個(gè)屬性： offset對(duì)應(yīng)類型：long MessageSize對(duì)應(yīng)類型：int32 data是message的具體內(nèi)容,Kafka的Message,Kafka的offset,每條消息在文件中的位置稱為offset（偏移量）。offset 為一個(gè)long型數(shù)字，它是唯一標(biāo)記一條消息。它唯一的標(biāo)記一條消息。kafka并沒有提供其他額外的索引機(jī)制來存儲(chǔ)offset，因?yàn)樵趉afka中幾 乎不允許對(duì)消息進(jìn)行“隨機(jī)讀寫”。 Partition中的每條Message由offset來表示它在

15、這個(gè)partition中的偏移量，這個(gè)offset不是該Message在partition數(shù)據(jù)文件中的實(shí)際存儲(chǔ)位置，而是邏輯上一個(gè)值，它唯一確定了partition中的一條Message。因此，可以認(rèn)為offset是partition中Message的id。,Kafka的 offset,怎樣記錄每個(gè)consumer處理的信息的狀態(tài)？在Kafka中僅保存了每個(gè)consumer已經(jīng)處理數(shù)據(jù)的offset。這樣有兩個(gè)好處： 1）保 存的數(shù)據(jù)量少 2）當(dāng)consumer出錯(cuò)時(shí)，重新啟動(dòng) consumer處理數(shù)據(jù)時(shí)，只需從最近的offset開始處理數(shù)據(jù)即可。,Kafka的消息處理機(jī)制,1. 發(fā)送到par

16、titions中的消息將會(huì)按照它接收的順序追加到日志中 2. 對(duì)于消費(fèi)者而言,它們消費(fèi)消息的順序和日志中消息順序一致. 3. 如果Topic的replication factor為N,那么允許N-1個(gè)kafka實(shí)例失效.,Kafka的消息處理機(jī)制,4. kafka對(duì)消息的重復(fù)、丟失、錯(cuò)誤以及順序型沒有嚴(yán)格的要求。 5. kafka提供at-least-once delivery,即當(dāng)consumer宕機(jī)后，有些消息可能會(huì)被重復(fù)delivery。 6. 因每個(gè)partition只會(huì)被consumergroup內(nèi)的一個(gè)consumer消費(fèi)，故kafka保證每個(gè)partition內(nèi)的消息會(huì)被順序的訂

17、閱。 7. Kafka為每條消息為每條消息計(jì)算CRC校驗(yàn)，用于錯(cuò)誤檢測，crc校驗(yàn)不通過的消息會(huì)直接被丟棄掉。 ack校驗(yàn)，當(dāng)消費(fèi)者消費(fèi)成功，返回ack信息！,數(shù)據(jù)傳輸?shù)氖聞?wù)定義,at most once: 最多一次,這個(gè)和JMS中非持久化消息類似.發(fā)送一次,無論成敗,將不會(huì)重發(fā). at least once: 消息至少發(fā)送一次,如果消息未能接受成功,可能會(huì)重發(fā),直到接收成功. exactly once: 消息只會(huì)發(fā)送一次. at most once: 消費(fèi)者fetch消息,然后保存offset,然后處理消息;當(dāng)client保存offset之后,但是在消息處理過程中出現(xiàn)了異常,導(dǎo)致部分消息未能

18、繼續(xù)處理.那么此后未處理的消息將不能被fetch到,這就是at most once. at least once: 消費(fèi)者fetch消息,然后處理消息,然后保存offset.如果消息處理成功之后,但是在保存offset階段zookeeper異常導(dǎo)致保存操作未能執(zhí)行成功,這就導(dǎo)致接下來再次fetch時(shí)可能獲得上次已經(jīng)處理過的消息,這就是at least once，原因offset沒有及時(shí)的提交給zookeeper，zookeeper恢復(fù)正常還是之前offset狀態(tài). exactly once: kafka中并沒有嚴(yán)格的去實(shí)現(xiàn)(基于2階段提交,事務(wù)),我們認(rèn)為這種策略在kafka中是沒有必要的.

19、注：通常情況下at-least-once是我們首選.(相比at most once而言,重復(fù)接收數(shù)據(jù)總比丟失數(shù)據(jù)要好).,Kafka的儲(chǔ)存策略,1. kafka以topic來進(jìn)行消息管理，每個(gè)topic包含多個(gè)part（ition），每個(gè)part對(duì)應(yīng)一個(gè)邏輯log，有多個(gè)segment組成。 2. 每個(gè)segment中存儲(chǔ)多條消息（見下圖），消息id由其邏輯位置決定，即從消息id可直接定位到消息的存儲(chǔ)位置，避免id到位置的額外映射。 3.broker 收到發(fā)布消息往對(duì)應(yīng) partition 的最后一個(gè) segment 上添加該消息，,Kafka的儲(chǔ)存策略,4. 每個(gè)part在內(nèi)存中對(duì)應(yīng)一個(gè)in

20、dex，記錄每個(gè)segment中的第一條消息偏移。 5. 發(fā)布者發(fā)到某個(gè)topic的 消息會(huì)被均勻的分布到多個(gè)part上（隨機(jī)或根據(jù)用戶指定的回調(diào)函數(shù)進(jìn)行分布），broker收到發(fā)布消息往對(duì)應(yīng)part的最后一個(gè)segment上添加 該消息，當(dāng)某個(gè)segment上的消息條數(shù)達(dá)到配置值或消息發(fā)布時(shí)間超過閾值時(shí)，segment上的消息會(huì)被flush到磁盤，只有flush到磁盤上的 消息訂閱者才能訂閱到，segment達(dá)到一定的大小后將不會(huì)再往該segment寫數(shù)據(jù)，broker會(huì)創(chuàng)建新的segment。,Kafka的數(shù)據(jù)傳輸,1. 發(fā)布者每次可發(fā)布多條消息（將消息加到一個(gè)消息集合中發(fā)布）， sub每

21、次迭代一條消息。 2. 不創(chuàng)建單獨(dú)的cache，使用系統(tǒng)的page cache。發(fā)布者順序發(fā)布，訂閱者通常比發(fā)布者滯后一點(diǎn)點(diǎn)，直接使用linux的page cache效果也比較后，同時(shí)減少了cache管理及垃圾收集的開銷。 3. 使用sendfile優(yōu)化網(wǎng)絡(luò)傳輸，減少一次內(nèi)存拷貝。,Kafka的消息發(fā)送的流程,由于 kafka broker 會(huì)持久化數(shù)據(jù)，broker 沒有內(nèi)存壓力，因此，consumer 非常適合采取 pull 的方式消費(fèi)數(shù)據(jù) Producer 向Kafka（push）推數(shù)據(jù) consumer 從kafka 拉（pull）數(shù)據(jù)。,kafka的消息發(fā)送的流程,消息處理的優(yōu)勢：

22、簡化 kafka 設(shè)計(jì) consumer 根據(jù)消費(fèi)能力自主控制消息拉取速度 consumer 根據(jù)自身情況自主選擇消費(fèi)模式，例如批量，重復(fù)消費(fèi)，從尾端開始消費(fèi)等 kafka 集群接收到 Producer 發(fā)過來的消息后，將其持久化到硬盤，并保留消息指定時(shí)長（可配置），而不關(guān)注消息是否被消費(fèi)。,Kafka設(shè)計(jì)原理實(shí)現(xiàn),直接使用 linux 文件系統(tǒng)的 cache，來高效緩存數(shù)據(jù)。 顯式分布式，即所有的 producer、broker 和 consumer 都會(huì)有多個(gè)，均為分布式的。Producer 和 broker 之間沒有負(fù)載均衡機(jī)制。broker 和 consumer 之間利用 zookee

23、per 進(jìn)行負(fù)載均衡。所有 broker 和 consumer 都會(huì)在 zookeeper 中進(jìn)行注冊(cè)，且 zookeeper 會(huì)保存他們的一些元數(shù)據(jù)信息。如果某個(gè) broker 和 consumer 發(fā)生了變化，所有其他的 broker 和 consumer 都會(huì)得到通知。,Kafka設(shè)計(jì)原理實(shí)現(xiàn),kafka 以 topic 來進(jìn)行消息管理，發(fā)布者發(fā)到某個(gè) topic 的消息會(huì)被均勻的分布到多個(gè) partition上 每個(gè) topic 包含多個(gè) partition，每個(gè) part 對(duì)應(yīng)一個(gè)邏輯 log，有多個(gè) segment 組成。 每個(gè) segment 中存儲(chǔ)多條消息，消息 id 由其邏輯

24、位置決定，即從消息 id 可直接定位到消息的存儲(chǔ)位置，避免 id 到位置的額外映射。 每個(gè) part 在內(nèi)存中對(duì)應(yīng)一個(gè) index，記錄每個(gè) segment 中的第一條消息偏移。 當(dāng)某個(gè) segment 上的消息條數(shù)達(dá)到配置值或消息發(fā)布時(shí)間超過閾值時(shí)，segment 上的消息會(huì)被 flush 到磁盤，只有 flush 到磁盤上的消息訂閱者才能訂閱到，segment 達(dá)到一定的大小后將不會(huì)再往該 segment 寫數(shù)據(jù)，broker 會(huì)創(chuàng)建新的 segment。,Kafka的通訊協(xié)議,Kafka的Producer、Broker和Consumer之間采用的是一套自行設(shè)計(jì)基于TCP層的協(xié)議，根據(jù)業(yè)務(wù)

25、需求定制，而非實(shí)現(xiàn)一套類似Protocol Buffer的通用協(xié)議。 基本數(shù)據(jù)類型： 定長數(shù)據(jù)類型：int8,int16,int32和int64，對(duì)應(yīng)到Java中就是byte, short, int和long。 變長數(shù)據(jù)類型：bytes和string。變長的數(shù)據(jù)類型由兩部分組成，分別是一個(gè)有符號(hào)整數(shù)N(表示內(nèi)容的長度)和N個(gè)字節(jié)的內(nèi)容。其中，N為-1表示內(nèi)容為null。bytes的長度由int32表示，string的長度由int16表示。 數(shù)組：數(shù)組由兩部分組成，分別是一個(gè)由int32類型的數(shù)字表示的數(shù)組長度N和N個(gè)元素。,Kafka的通訊協(xié)議,Kafka通訊的基本單位是Request/Res

26、ponse 基本結(jié)構(gòu)： RequestOrResponse = MessageSize (RequestMessage | ResponseMessage) 通訊過程： 客戶端打開與服務(wù)器端的Socket 往Socket寫入一個(gè)int32的數(shù)字(數(shù)字表示這次發(fā)送的Request有多少字節(jié)) 服務(wù)器端先讀出一個(gè)int32的整數(shù)從而獲取這次Request的大小 然后讀取對(duì)應(yīng)字節(jié)數(shù)的數(shù)據(jù)從而得到Request的具體內(nèi)容 服務(wù)器端處理了請(qǐng)求后，也用同樣的方式來發(fā)送響應(yīng)。,Kafka的通訊協(xié)議,RequestMessage結(jié)構(gòu): RequestMessage = ApiKey ApiVersion Co

27、rrelationId ClientId Request,Kafka的通訊協(xié)議,ResponseMessage結(jié)構(gòu): ResponseMessage = CorrelationId Response Kafka采用是經(jīng)典的Reactor(同步IO)模式，也就是1個(gè)Acceptor響應(yīng)客戶端的連接請(qǐng)求，N個(gè)Processor來讀取數(shù)據(jù)，這種模式可以構(gòu)建出高 性能的服務(wù)器。,Kafka的通訊協(xié)議,Message:Producer生產(chǎn)的消息,鍵-值對(duì) Message = Crc MagicByte Attributes Key Value,Kafka的通訊協(xié)議,MessageSet:用來組合多條Me

28、ssage，它在每條Message的基礎(chǔ)上加上了Offset和MessageSize MessageSet = Offset MessageSize Message,Kafka的通訊協(xié)議組件關(guān)系,Request/Respone和Message/MessageSet的關(guān)系： 備注：Kafka的通訊協(xié)議中不含Schema，格式也比較簡單，這樣設(shè)計(jì)的好處是協(xié)議自身的Overhead小，再加上把多條Message放在一起做壓縮，提高壓縮比率，從而在網(wǎng)絡(luò)上傳輸?shù)臄?shù)據(jù)量會(huì)少一些。,Kafka的分布式實(shí)現(xiàn),一個(gè)Topic的多個(gè)partitions,被分布在kafka集群中的多個(gè)server上;每個(gè)serve

29、r(kafka實(shí)例)負(fù)責(zé)partitions中消息的讀寫操作; 此外kafka還可以配置partitions需要備份的個(gè)數(shù)(replicas),每個(gè)partition將會(huì)被備份到多臺(tái)機(jī)器上,以提高可用性; 基于replicated方案,那么就意味著需要對(duì)多個(gè)備份進(jìn)行調(diào)度; 每個(gè)partition都有一個(gè)server為leader;leader負(fù)責(zé)所有的讀寫操作,如果leader失效,那么將會(huì)有其他follower來接管(成為新的leader); follower只是單調(diào)的和leader跟進(jìn),同步消息即可.由此可見作為leader的server承載了全部的請(qǐng)求壓力,因此從集群的整體考慮,有多少個(gè)

30、partitions就意味著有多少個(gè)leader; kafka會(huì)將leader均衡的分散在每個(gè)實(shí)例上,來確保整體的性能穩(wěn)定.,Kafka數(shù)據(jù)持久化,數(shù)據(jù)持久化： 發(fā)現(xiàn)線性的訪問磁盤，很多時(shí)候比隨機(jī)的內(nèi)存訪問快得多 傳統(tǒng)的使用內(nèi)存做為磁盤的緩存 Kafka直接將數(shù)據(jù)寫入到日志文件中 日志數(shù)據(jù)持久化特性： 寫操作：通過將數(shù)據(jù)追加到文件中實(shí)現(xiàn) 讀操作：讀的時(shí)候從文件中讀就好了 對(duì)比JVM特性： Java對(duì)象占用空間是非常大的，差不多是要存儲(chǔ)的數(shù)據(jù)的兩倍甚至更高 隨著堆中數(shù)據(jù)量的增加，垃圾回收回變的越來越困難 優(yōu)勢：讀操作不會(huì)阻塞寫操作和其他操作，數(shù)據(jù)大小不對(duì)性能產(chǎn)生影響； 沒有容量限制（相對(duì)于內(nèi)存來

31、說）的硬盤空間建立消息系統(tǒng)； 線性訪問磁盤，速度快，可以保存任意一段時(shí)間！,Kafka安裝,下載 /downloads.html 解壓 tar -zxvf kafka_2.10-.tgz 啟動(dòng)服務(wù) 首先啟動(dòng)zookeeper服務(wù) bin/zookeeper-server-start.sh config/perties 啟動(dòng)Kafka bin/kafka-server-start.sh config/perties /dev/null 2&1 & 創(chuàng)建topic 創(chuàng)建一個(gè)test的topic，

32、一個(gè)分區(qū)一個(gè)副本 bin/kafka-topics.sh -create -zookeeper localhost:2181 -replication-factor 1 -partitions 1 -topic test 查看主題 bin/kafka-topics.sh -list -zookeeper localhost:2181 查看主題詳情 bin/kafka-topics.sh -describe -zookeeper localhost:2181 -topic test 刪除主題 bin/kafka-topics.sh -zookeeper localhost:2181 -delet

33、e -topic test,Kafka客戶端操作,創(chuàng)建生產(chǎn)者 producer bin/kafka-console-producer.sh -broker-list localhost:9092 -topic test 創(chuàng)建消費(fèi)者 consumer bin/kafka-console-consumer.sh -zookeeper localhost:2181 -topic test -from-beginning 參數(shù)使用幫組信息查看： 生產(chǎn)者參數(shù)查看：bin/kafka-console-producer.sh 消費(fèi)者參數(shù)查看：bin/kafka-console-consumer.sh,Kaf

34、ka多broker部署,修改config/perties broker.id=0 port=9020 log.dirs=/tmp/kafka0-logs 復(fù)制perties生成perties broker.id=1#id不能一樣 port=9040#port不能一樣 log.dirs=/tmp/kafka1-logs 啟動(dòng)多個(gè)broker bin/kafka-server-start.sh config/perties & bin/kafka-server-start.sh config/service1

35、.properties & 創(chuàng)建主題 bin/kafka-topics.sh -create -zookeeper localhost:2181 -replication-factor 3 -partitions 1 -topic test,kafka集群安裝,安裝zk集群 修改配置文件 broker.id： 唯一，填數(shù)字 ：唯一，填服務(wù)器 zookeeper.connect=34:2181,32:2181,33:2181,Kafka的核心配置,perties 配置詳情見注釋 brok

36、er.id=0 work.threads=2 num.io.threads=8 socket.send.buffer.bytes=1048576 socket.receive.buffer.bytes=1048576 socket.request.max.bytes=104857600 log.dirs=/tmp/kafka-logs num.partitions=2 log.retention.hours=168 log.segment.bytes=536870912 erval.ms=60000 log.cleaner.enable=false

37、 zookeeper.connect=localhost:2181 zookeeper.connection.timeout.ms=1000000,Kafka的一致性,MQ要實(shí)現(xiàn)從producer到consumer之間的可靠的消息傳送和分發(fā)。傳統(tǒng)的MQ系統(tǒng)通常都是通過broker和consumer間的確認(rèn) （ack）機(jī)制實(shí)現(xiàn)的，并在broker保存消息分發(fā)的狀態(tài)。即使這樣一致性也是很難保證的（當(dāng)然kafka也支持ack）。 kafka保證一致性的做法是由 consumer自己保存狀態(tài)，也不要任何確認(rèn)。這樣雖然consumer負(fù)擔(dān)更重，但其實(shí)更靈活了。因?yàn)椴还躢onsumer上任何原因?qū)е滦枰?/p>

38、新處 理消息，都可以再次從broker獲得。,Kafka的高可用性,Kafaka可以將log文件復(fù)制到其他topic的分隔點(diǎn)(可以看成是server)。當(dāng)一個(gè)server在集群中fails，可以允許自動(dòng)的failover到其他的復(fù)制的server，所以消息可以繼續(xù)存在在這種情況下。,Kafka的zero-copy,采用 linux Zero-Copy 提高發(fā)送性能。傳統(tǒng)的數(shù)據(jù)發(fā)送需要發(fā)送 4 次上下文切換，采用 sendfile 系統(tǒng)調(diào)用之后，數(shù)據(jù)直接在內(nèi)核態(tài)交換，系統(tǒng)上下文切換減少為 2 次。根據(jù)測試結(jié)果，可以提高 60% 的數(shù)據(jù)發(fā)送性能。,Kafka的zero-copy,在Kafka上，有

39、兩個(gè)原因可能導(dǎo)致低效：1）太多的網(wǎng)絡(luò)請(qǐng)求 2）過多的字節(jié)拷貝。為了提高效率，Kafka把message分成一組一組的，每次請(qǐng)求會(huì)把一組message發(fā)給相應(yīng)的consumer。 此外， 為了減少字節(jié)拷貝，采用了sendfile系統(tǒng)調(diào)用。為了理解sendfile原理，先說一下傳統(tǒng)的利用socket發(fā)送文件要進(jìn)行拷貝,Sendfile系統(tǒng)調(diào)用,Kafka的負(fù)載均衡,Producer和broker之間沒有負(fù)載均衡機(jī)制。 負(fù)載均衡可以分為兩個(gè)部分：producer發(fā)消息的負(fù)載均衡和consumer讀消息的負(fù)載均衡。 producer有一個(gè)到當(dāng)前所有broker的連接池，當(dāng)一個(gè)消息需要發(fā)送時(shí)，需要決定發(fā)

40、到哪個(gè)broker（即partition）。 consumer讀取消息時(shí)，除了考慮當(dāng)前的broker情況外，還要考慮其他consumer的情況，才能決定從哪個(gè)partition讀取消息。 多個(gè) partition 需要選取出 lead partition，lead partition 負(fù)責(zé)讀寫，broker和consumer之間利用zookeeper進(jìn)行負(fù)載均衡。 所有broker和consumer都會(huì)在zookeeper中進(jìn)行注冊(cè)，且 zookeeper會(huì)保存他們的一些元數(shù)據(jù)信息。如果某個(gè)broker和consumer發(fā)生了變化，所有其他的broker和consumer都會(huì)得到 通知。,Ka

41、fka 可擴(kuò)展性,當(dāng)需要增加 broker 結(jié)點(diǎn)時(shí)，新增的 broker 會(huì)向 zookeeper 注冊(cè)，而 producer 及 consumer 會(huì)根據(jù)注冊(cè)在 zookeeper 上的 watcher 感知這些變化，并及時(shí)作出調(diào)整，這樣就保證了添加或去除broker時(shí)，各broker間仍能自動(dòng)實(shí)現(xiàn)負(fù)載均衡。,Kafka的Zookeeper協(xié)調(diào)控制,1. 管理broker與consumer的動(dòng)態(tài)加入與離開。 2. 觸發(fā)負(fù)載均衡，當(dāng)broker或consumer加入或離開時(shí)會(huì)觸發(fā)負(fù)載均衡算法，使得一個(gè)consumer group內(nèi)的多個(gè)consumer的訂閱負(fù)載平衡。 3. 維護(hù)消費(fèi)關(guān)系及每個(gè)

42、partion的消費(fèi)信息。 Zookeeper上的細(xì)節(jié)： 1. 每個(gè)broker啟動(dòng)后會(huì)在zookeeper上注冊(cè)一個(gè)臨時(shí)的broker registry，包含broker的ip地址和端口號(hào)，所存儲(chǔ)的topics和partitions信息。 2. 每個(gè)consumer啟動(dòng)后會(huì)在zookeeper上注冊(cè)一個(gè)臨時(shí)的consumer registry：包含consumer所屬的consumer group以及訂閱的topics。 3. 每個(gè)consumer group關(guān) 聯(lián)一個(gè)臨時(shí)的owner registry和一個(gè)持久的offset registry。對(duì)于被訂閱的每個(gè)partition包含一個(gè)owner registry，內(nèi)容為訂閱這個(gè)partition的consumer id；同時(shí)包含一個(gè)offset registry，內(nèi)