培訓(xùn)課程Kafka

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

29、多個partitions,被分布在kafka集群中的多個server上;每個server(kafka實例)負(fù)責(zé)partitions中消息的讀寫操作; 此外kafka還可以配置partitions需要備份的個數(shù)(replicas),每個partition將會被備份到多臺機器上,以提高可用性; 基于replicated方案,那么就意味著需要對多個備份進行調(diào)度; 每個partition都有一個server為leader;leader負(fù)責(zé)所有的讀寫操作,如果leader失效,那么將會有其他follower來接管(成為新的leader); follower只是單調(diào)的和leader跟進,同步消息即可.由此

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

31、寫操作和其他操作，數(shù)據(jù)大小不對性能產(chǎn)生影響； 沒有容量限制（相對于內(nèi)存來說）的硬盤空間建立消息系統(tǒng)； 線性訪問磁盤，速度快，可以保存任意一段時間！Kafka安裝安裝下載解壓tar -zxv啟動服務(wù)首先啟動zookeeper服務(wù)啟動Kafkabin/kafka-server-start.sh config/perties /dev/null 2&1 &創(chuàng)建topic創(chuàng)建一個test的topic，一個分區(qū)一個副本bin/kafka-topics.sh -create -zookeeper localhost:2181 -replication-factor 1

32、-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 -delete -topic testKafka客戶端操作客戶端操作創(chuàng)建生產(chǎn)者 producerbin/kafka-console-producer.sh -broker-list localh

33、ost:9092 -topic test 創(chuàng)建消費者 consumerbin/kafka-console-consumer.sh -zookeeper localhost:2181 -topic test -from-beginning參數(shù)使用幫組信息查看：Kafka多多broker部署部署broker.id=0port=9020log.dirs=/tmp/pertiesbroker.id=1#id不能一樣port=9040#port不能一樣log.dirs=/tmp/kafka1-logs啟動多個brokerbin/kafka-server-st

34、art.sh config/perties &bin/kafka-server-start.sh config/perties &創(chuàng)建主題bin/kafka-topics.sh -create -zookeeper localhost:2181 -replication-factor 3 -partitions 1 -topic testkafka集群安裝集群安裝安裝zk集群修改配置文件broker.id： 唯一，填數(shù)字：唯一，填服務(wù)器zookeeper.connect=34:2181,19

35、32:2181,33:2181Kafka的核心配置的核心配置s配置詳情見注釋broker.id=work.threads=2num.io.threads=8socket.send.buffer.bytes=1048576socket.receive.buffer.bytes=1048576socket.request.max.bytes=104857600log.dirs=/tmp/kafka-logsnum.partitions=2log.retention.hours=168log.segment.bytes=536870912log.reten

36、erval.ms=60000log.cleaner.enable=falsezookeeper.connect=localhost:2181zookeeper.connection.timeout.ms=1000000Kafka的一致性的一致性MQ要實現(xiàn)從producer到consumer之間的可靠的消息傳送和分發(fā)。傳統(tǒng)的MQ系統(tǒng)通常都是通過broker和consumer間的確認(rèn) （ack）機制實現(xiàn)的，并在broker保存消息分發(fā)的狀態(tài)。即使這樣一致性也是很難保證的（當(dāng)然kafka也支持ack）。kafka保證一致性的做法是由 consumer自己保存狀態(tài)，也不要任

37、何確認(rèn)。這樣雖然consumer負(fù)擔(dān)更重，但其實更靈活了。因為不管consumer上任何原因?qū)е滦枰匦绿?理消息，都可以再次從broker獲得。 Kafka的高可用性的高可用性Kafaka可以將log文件復(fù)制到其他topic的分隔點(可以看成是server)。當(dāng)一個server在集群中fails，可以允許自動的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)上下文切換減少為

38、2 次。根據(jù)測試結(jié)果，可以提高 60% 的數(shù)據(jù)發(fā)送性能。Kafka的的zero-copy 在Kafka上，有兩個原因可能導(dǎo)致低效：1）太多的網(wǎng)絡(luò)請求 2）過多的字節(jié)拷貝。為了提高效率，Kafka把message分成一組一組的，每次請求會把一組message發(fā)給相應(yīng)的consumer。 此外， 為了減少字節(jié)拷貝，采用了sendfile系統(tǒng)調(diào)用。為了理解sendfile原理，先說一下傳統(tǒng)的利用socket發(fā)送文件要進行拷貝Sendfile系統(tǒng)調(diào)用Kafka的負(fù)載均衡的負(fù)載均衡Producer和broker之間沒有負(fù)載均衡機制。 負(fù)載均衡可以分為兩個部分：producer發(fā)消息的負(fù)載均衡和consu

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

40、broker和consumer發(fā)生了變化，所有其他的broker和consumer都會得到 通知。Kafka 可擴展性可擴展性當(dāng)需要增加 broker 結(jié)點時，新增的 broker 會向 zookeeper 注冊，而 producer 及 consumer 會根據(jù)注冊在 zookeeper 上的 watcher 感知這些變化，并及時作出調(diào)整，這樣就保證了添加或去除broker時，各broker間仍能自動實現(xiàn)負(fù)載均衡。Kafka的的Zookeeper協(xié)調(diào)控制協(xié)調(diào)控制1. 管理broker與consumer的動態(tài)加入與離開。2. 觸發(fā)負(fù)載均衡，當(dāng)broker或consumer加入或離開時會觸發(fā)負(fù)載

41、均衡算法，使得一個consumer group內(nèi)的多個consumer的訂閱負(fù)載平衡。3. 維護消費關(guān)系及每個partion的消費信息。Zookeeper上的細(xì)節(jié)：1. 每個broker啟動后會在zookeeper上注冊一個臨時的broker registry，包含broker的ip地址和端口號，所存儲的topics和partitions信息。2. 每個consumer啟動后會在zookeeper上注冊一個臨時的consumer registry：包含consumer所屬的consumer group以及訂閱的topics。3. 每個consumer group關(guān) 聯(lián)一個臨時的owner registry和一個持久的offset registry。對于被訂閱的每個partition包含一個owner registry，內(nèi)容為訂閱這個partition的consumer id；同時包含一個offset registry，內(nèi)容為上一次訂閱的offset。kafka java操作操作生產(chǎn)者消費者pom依賴org.apache.kafkakafka_2.100.8