培訓課程Kafka.ppt

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

32、一個分區(qū)一個副本 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)建生產者 producer bin/kafka-console-producer.sh -broker-list localhost:9092 -topic test 創(chuàng)建消費者 consumer bin/kafka-console-consumer.sh -zookeeper localhost:2181 -topic test -from-beginning 參數(shù)使用幫組信息查看： 生產者參數(shù)查看：bin/kafka-console-producer.sh 消費者參數(shù)查看：bin/kafka-console-consumer.sh,Kaf

34、ka多broker部署,修改config/perties broker.id=0 port=9020 log.dirs=/tmp/kafka0-logs 復制perties生成perties broker.id=1#id不能一樣 port=9040#port不能一樣 log.dirs=/tmp/kafka1-logs 啟動多個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ù)字 ：唯一，填服務器 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要實現(xiàn)從producer到consumer之間的可靠的消息傳送和分發(fā)。傳統(tǒng)的MQ系統(tǒng)通常都是通過broker和consumer間的確認 （ack）機制實現(xiàn)的，并在broker保存消息分發(fā)的狀態(tài)。即使這樣一致性也是很難保證的（當然kafka也支持ack）。 kafka保證一致性的做法是由 consumer自己保存狀態(tài)，也不要任何確認。這樣雖然consumer負擔更重，但其實更靈活了。因為不管consumer上任何原因導致需要重

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

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

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

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

42、partion的消費信息。 Zookeeper上的細節(jié)： 1. 每個broker啟動后會在zookeeper上注冊一個臨時的broker registry，包含broker的ip地址和端口號，所存儲的topics和partitions信息。 2. 每個consumer啟動后會在zookeeper上注冊一個臨時的consumer registry：包含consumer所屬的consumer group以及訂閱的topics。 3. 每個consumer group關 聯(lián)一個臨時的owner registry和一個持久的offset registry。對于被訂閱的每個partition包含一個owner registry，內容為訂閱這個partition的consumer id；同時包含一個offset registry，內