1- 學習目標

• kafka是什么？能做什么？有什么特點？
• kafka作業原理
• 安裝部署和常用命令串列
• 和其他訊息中間件對比
• 碼上牛掰大佬的深度解讀 https://blog.csdn.net/u013256816/article/details/71091774 ，需要反復看

2- 簡介

2.1 是什么

Apache Kafka是基于發布訂閱的容錯訊息系統，

• 什么是訊息系統？

訊息系統負責將資料從一個應用程式傳輸到另一個應用程式，因此應用程式可以專注于資料，但不擔心如何共享它， 分布式訊息傳遞基于可靠訊息佇列的概念， 訊息在客戶端應用程式和訊息傳遞系統之間異步排隊， 有兩種型別的訊息模式可用: 一種是點對點，另一種是發布訂閱(pub-sub)訊息系統，

• 點對點訊息系統

在點對點系統中，訊息被保留在佇列中， 一個或多個消費者可以消耗佇列中的訊息，但是特定訊息只能由最多一個消費者消費， 一旦消費者讀取佇列中的訊息，它就從該佇列中消失， 該系統的典型示例是訂單處理系統，其中每個訂單將由一個訂單處理器處理，但多個訂單處理器也可以同時作業，

• 發布訂閱訊息系統

在發布 - 訂閱系統中，訊息被保留在主題中， 與點對點系統不同，消費者可以訂閱一個或多個主題并使用該主題中的所有訊息， 在發布 - 訂閱系統中，訊息生產者稱為發布者，訊息使用者稱為訂閱者， 一個現實生活的例子是Dish電視，它發布不同的渠道，如運動，電影，音樂等，任何人都可以訂閱自己的頻道集，并獲得他們訂閱的頻道時可用，

2.2 優點

可靠性 - Kafka是分布式，磁區，復制和容錯的，資料流磁區(Partition)存盤在多個機器上，
可擴展性 - Kafka訊息傳遞系統輕松縮放，無需停機，
耐用性 - Kafka使用分布式提交日志，這意味著訊息會盡可能快地保留在磁盤上，因此它是持久的，
性能 - Kafka對于發布和訂閱訊息都具有高吞吐量， 即使存盤了許多TB的訊息，它也保持穩定的性能，

2.3 常用場景

指標 - Kafka通常用于操作監控資料， 這涉及聚合來自分布式應用程式的統計資訊，以產生操作資料的集中饋送，
日志聚合解決方案 - Kafka可用于跨組織從多個服務收集日志，并使它們以標準格式提供給多個服務器，
流處理 - 流行的框架(如Storm和Spark Streaming)從主題中讀取資料，對其進行處理，并將處理后的資料寫入新主題，供用戶和應用程式使用， Kafka的強耐久性在流處理的背景關系中也非常有用，

2.4 為什么使用訊息佇列

2.4.1 解耦

看這么個場景，A 系統發送資料到 BCD 三個系統，通過介面呼叫發送，如果 E 系統也要這個資料呢？那如果 C 系統現在不需要了呢？A 系統負責人幾乎崩潰…

在這個場景中，A 系統跟其它各種亂七八糟的系統嚴重耦合，A 系統產生一條比較關鍵的資料，很多系統都需要 A 系統將這個資料發送過來，A 系統要時時刻刻考慮 BCDE 四個系統如果掛了該咋辦？要不要重發，要不要把訊息存起來？頭發都白了啊！

如果使用 MQ，A 系統產生一條資料，發送到 MQ 里面去，哪個系統需要資料自己去 MQ 里面消費，如果新系統需要資料，直接從 MQ 里消費即可；如果某個系統不需要這條資料了，就取消對 MQ 訊息的消費即可，這樣下來，A 系統壓根兒不需要去考慮要給誰發送資料，不需要維護這個代碼，也不需要考慮人家是否呼叫成功、失敗超時等情況，

總結：通過一個 MQ，Pub/Sub 發布訂閱訊息這么一個模型，A 系統就跟其它系統徹底解耦了，

2.4.2 異步

再來看一個場景，A 系統接收一個請求，需要在自己本地寫庫，還需要在 BCD 三個系統寫庫，自己本地寫庫要 3ms，BCD 三個系統分別寫庫要 300ms、450ms、200ms，最終請求總延時是 3 + 300 + 450 + 200 = 953ms，接近 1s，用戶感覺搞個什么東西，慢死了慢死了，用戶通過瀏覽器發起請求，等待個 1s，這幾乎是不可接受的，

一般互聯網類的企業，對于用戶直接的操作，一般要求是每個請求都必須在 200 ms 以內完成，對用戶幾乎是無感知的，

如果使用 MQ，那么 A 系統連續發送 3 條訊息到 MQ 佇列中，假如耗時 5ms，A 系統從接受一個請求到回傳回應給用戶，總時長是 3 + 5 = 8ms，對于用戶而言，其實感覺上就是點個按鈕，8ms 以后就直接回傳了，爽！網站做得真好，真快！

2.4.3 削峰

每天 0:00 到 12:00，A 系統風平浪靜，每秒并發請求數量就 50 個，結果每次一到 12:00 ~ 13:00 ，每秒并發請求數量突然會暴增到 5k+ 條，但是系統是直接基于 MySQL 的，大量的請求涌入 MySQL，每秒鐘對 MySQL 執行約 5k 條 SQL，

一般的 MySQL，扛到每秒 2k 個請求就差不多了，如果每秒請求到 5k 的話，可能就直接把 MySQL 給打死了，導致系統崩潰，用戶也就沒法再使用系統了，

但是高峰期一過，到了下午的時候，就成了低峰期，可能也就 1w 的用戶同時在網站上操作，每秒中的請求數量可能也就 50 個請求，對整個系統幾乎沒有任何的壓力，

如果使用 MQ，每秒 5k 個請求寫入 MQ，A 系統每秒鐘最多處理 2k 個請求，因為 MySQL 每秒鐘最多處理 2k 個，A 系統從 MQ 中慢慢拉取請求，每秒鐘就拉取 2k 個請求，不要超過自己每秒能處理的最大請求數量就 ok，這樣下來，哪怕是高峰期的時候，A 系統也絕對不會掛掉，而 MQ 每秒鐘 5k 個請求進來，就 2k 個請求出去，結果就導致在中午高峰期（1 個小時），可能有幾十萬甚至幾百萬的請求積壓在 MQ 中，

這個短暫的高峰期積壓是 ok 的，因為高峰期過了之后，每秒鐘就 50 個請求進 MQ，但是 A 系統依然會按照每秒 2k 個請求的速度在處理，所以說，只要高峰期一過，A 系統就會快速將積壓的訊息給解決掉，

2.4.4 訊息佇列的缺點

• 系統可用性降低

MQ 一掛，整套系統崩潰

• 系統復雜度提高

怎么保證訊息沒有重復消費？怎么處理訊息丟失的情況？怎么保證訊息傳遞的順序性？

• 一致性問題

A 系統處理完了直接回傳成功了，人都以為你這個請求就成功了；但是問題是，要是 BCD 三個系統那里，BD 兩個系統寫庫成功了，結果 C 系統寫庫失敗了，咋整？你這資料就不一致了，

3- 作業原理

3.1- 術語

• 生產者和消費者（producer和consumer）

訊息的發送者叫 Producer，訊息的使用者和接受者是 Consumer，生產者采用推（push）模式將訊息發布到broker，每條訊息都被追加（append）到磁區（patition）中，屬于順序寫磁盤（順序寫磁盤效率比隨機寫記憶體要高，保障kafka吞吐率），消費者采用拉取（pull）方式獲取訊息并進一步處理，消費者自己記錄消費狀態，每個消費者互相獨立地順序讀取每個磁區的訊息，

• broker

Kafka 集群中有很多臺 Server，其中每一臺 Server 都可以存盤訊息，將每一臺 Server 稱為一個 kafka 實體，也叫做 broker，

• 主題（topic）

一個 topic 里保存的是同一類訊息，相當于對訊息的分類，每個 producer 將訊息發送到 kafka 中，都需要指明要存的 topic 是哪個，也就是指明這個訊息屬于哪一類，

• 磁區（partition）

每個 topic 都可以分成多個 partition，每個 partition 在存盤層面是 append log 檔案，任何發布到此 partition 的訊息都會被直接追加到 log 檔案的尾部，為什么要進行磁區呢？最根本的原因就是：kafka基于檔案進行存盤，當檔案內容大到一定程度時，很容易達到單個磁盤的上限，因此，采用磁區的辦法，一個磁區對應一個檔案，這樣就可以將資料分別存盤到不同的server上去，另外這樣做也可以負載均衡，容納更多的消費者，

• 偏移量（Offset）

一個磁區對應一個磁盤上的檔案，而訊息在檔案中的位置就稱為 offset（偏移量），offset 為一個 long 型數字，它可以唯一標記一條訊息，由于kafka 并沒有提供其他額外的索引機制來存盤 offset，檔案只能順序的讀寫，所以在kafka中幾乎不允許對訊息進行“隨機讀寫”，

• 副本（replication）

kafka 還可以配置 partitions 需要備份的個數(replicas),每個 partition 將會被備份到多臺機器上,以提高可用性，備份的數量可以通過組態檔指定，同一個partition可能會有多個replication（對應 server.properties 配置中的 default.replication.factor=N），沒有replication的情況下，一旦broker 宕機，其上所有 patition 的資料都不可被消費，同時producer也不能再將資料存于其上的patition，

這種冗余備份的方式在分布式系統中是很常見的，那么既然有副本，就涉及到對同一個檔案的多個備份如何進行管理和調度，kafka 采取的方案是：每個 partition 選舉一個 server 作為“leader”，由 leader 負責所有對該磁區的讀寫，其他 server 作為 follower 只需要簡單的與 leader 同步，保持跟進即可，如果原來的 leader 失效，會重新選舉由其他的 follower 來成為新的 leader，

至于如何選取 leader，實際上如果我們了解 ZooKeeper，就會發現其實這正是 Zookeeper 所擅長的，Kafka 使用 ZK 在 Broker 中選出一個 Controller，用于 Partition 分配和 Leader 選舉，

另外，這里我們可以看到，實際上作為 leader 的 server 承擔了該磁區所有的讀寫請求，因此其壓力是比較大的，從整體考慮，從多少個 partition 就意味著會有多少個leader，kafka 會將 leader 分散到不同的 broker 上，確保整體的負載均衡，

• OSR（Out of-Sync Replicas）

即與Leader資料不一致的副本串列，

• ISR

In-sync Replicas（副本同步串列），ISR中的副本都是與Leader資料一致的副本，不在ISR中的Follower副本是與Leader資料不一致的，Leader副本天然就在 ISR 中，但ISR不只是Follower副本集合，是一個動態調整的集合，由Leader負責維護，如果一個follower的同步時間過長（大于閾值），就會被ISR剔除，

• AR（Assigned Replicas）

即已分配的副本串列，是指某個Partition的所有副本，AR = ISR + OSR

• HW

俗稱高水位，HighWatermark的縮寫，具體含義看下圖：

3.2- 集群組件

• Broker（代理）

Broker就是kafka實體，即Kafka集群通常由多個Broker組成以保持負載平衡， Broker是無狀態的，所以他們使用ZooKeeper來維護它們的集群狀態， 一個Broker可以每秒處理數十萬次讀取和寫入，每個Broker可以處理TB的訊息，而沒有性能影響， Kafka經紀人領導選舉可以由ZooKeeper完成，

• ZooKeeper

ZooKeeper用于管理和協調Broker， ZooKeeper服務主要用于通知生產者和消費者Kafka系統中存在任何新Broker或Kafka系統中Broker失敗， 根據Zookeeper接收到關于Broker的存在或失敗的通知，然后生產者和消費者采取決定并開始與某些其他Broker協調他們的任務，Kafka實體通過 Zookeeper 集群共享資訊，Kafka實體 在 Zookeeper 中存盤基本元資料，例如關于主題，代理，消費者偏移(佇列讀取器)等的資訊，

Producers（生產者）
生產者將資料推送給Broker， 當新Broker啟動時，所有生產者搜索它并自動向該新Broker發送訊息， Kafka生產者不等待來自Broker的確認，并且發送訊息的速度與Broker可以處理的一樣快，producer不需要在zookeeper中注冊，

• Consumers（消費者）

因為Broker是無狀態的，這意味著消費者必須通過使用磁區偏移來維護已經消耗了多少訊息， 如果消費者確認特定的訊息偏移，則意味著消費者已經消費了所有先前的訊息， 消費者向Broker發出異步拉取請求，以具有準備好消耗的位元組緩沖區， 消費者可以簡單地通過提供偏移值來快退或跳到磁區中的任何點， 消費者偏移值由ZooKeeper通知，消費者需要在zookeeper中注冊，

• ConsumerGroup

每個Consumer屬于一個特定的Consumer Group，一條訊息可以發送到多個不同的Consumer Group，但是一個Consumer Group中只能有一個Consumer能夠消費該訊息

3.3- 作業流程

• 發布訂閱訊息的作業流程

1. 生產者定期向主題發送訊息，
2. Broker存盤該特定主題配置的磁區中的所有訊息， 它確保訊息在磁區之間平等共享， 如果生產者發送兩個訊息并且有兩個磁區，Broker 將在第一磁區中存盤一個訊息，在第二磁區中存盤第二訊息，
3. 消費者訂閱特定主題，
4. 一旦消費者訂閱主題，Kafka 將向消費者提供主題的當前偏移，并且還將偏移保存在 Zookeeper 系統中，
5. 消費者將定期請求 Kafka (如100 Ms)新訊息，
6. 一旦 Kafka 收到來自生產者的訊息，它將這些訊息轉發給消費者，
7. 消費者將收到訊息并進行處理，
8. 一旦訊息被處理，消費者將向 Kafka 代理發送確認，
9. 一旦 Kafka 收到確認，它將偏移更改為新值，并在 Zookeeper 中更新它，
10. 以上流程將重復，直到消費者停止請求，
11. 消費者可以隨時回退/跳到所需的主題偏移量，并閱讀所有后續訊息，

• 佇列訊息/用戶組的作業流

訂閱具有相同 Group ID 的主題的消費者被認為是單個組，并且訊息在它們之間共享，作業流程如下：

1. 生產者以固定間隔向某個主題發送訊息，
2. Broker存盤該特定主題配置的磁區中的所有訊息，
3. 單個消費者訂閱特定主題，假設 Topic-01 為 Group ID 為 Group-1 ，
4. Kafka 與消費者互動，直到新消費者以相同的組 ID 訂閱相同主題Topic-01 ，
5. 一旦新消費者到達，Kafka 將其操作切換到共享模式，并在兩個消費者之間共享資料， 此共享將繼續，直到用戶數達到為該特定主題配置的磁區數，
6. 一旦消費者的數量超過磁區的數量，新消費者將不會接收任何進一步的訊息，直到現有消費者取消訂閱任何一個消費者， 出現這種情況是因為 Kafka 中的每個消費者將被分配至少一個磁區，并且一旦所有磁區被分配給現有消費者，新消費者將必須等待，

• 生產者生產訊息流程

將訊息發送給Broker并形成可供消費者消費的訊息資料的程序如下：

1. Producer先從Zookeeper中找到Partition的Leader
2. Producer將訊息發送給磁區的Leader，
3. Leader將訊息接入本地的Log，并通知ISR的Followers，
4. ISR中的Followers從Leader中pull訊息，寫入本地Log后向Leader發送ACK，
5. Leader收到所有ISR中的Followers的ACK后，增加HW并向Producer發送ACK，表示訊息寫入成功，

• consumer 作業流程

1. 連接 ZK 集群，拿到對應 topic 的 partition 資訊和 partition 的 leader 的相關資訊
2. 連接到對應 leader 對應的 broker
3. consumer 將自己保存的 offset 發送給 leader
4. leader 根據 offset 等資訊定位到 segment（索引檔案和日志檔案）
5. 根據索引檔案中的內容，定位到日志檔案中該偏移量對應的開始位置讀取相應長度的資料并回傳給 consumer

• 訊息路由策略

在通過API方式發布訊息時，生產者是以Record為訊息進行發布的，Record中包含key與value，value才是訊息本身，而key用于路由訊息所要存放Partition，訊息要寫入到哪個Partition并不是隨機的，而是由路由策略決定，

1. 如果指定了Partition，則直接寫入到指定的Partition，
2. 如果沒有指定Partition但指定了key，則通過對key的hash值與Partition數量取模，結果就是要選出的Partition索引，
3. 如果Partition和key都未指定，則使用輪詢演算法選出一個Partition，
4. 增加磁區時，Partition內的訊息不會重新進行分配，隨著資料繼續寫入，新磁區才會參與再平衡

• Rebalance

Rebalance本質上是一種協議，規定了一個Consumer Group下的所有Consumer如何達成一致，來分配訂閱Topic的每個磁區，如某個Group下有20個Consumer實體，訂閱了一個具有100個磁區的Topic，正常情況下，Kafka平均會為每個Consumer分配5個磁區，磁區的分配的程序就叫Rebalance，Consumer Group觸發Rebalance的條件有 3 個：

1. 組成員數發生變更，如有新的Consumer實體加入組或者離開組，或有Consumer實體崩潰從組中洗掉，
2. 訂閱Topic數發生變更，Consumer Group可以使用正則運算式的方式訂閱主題，在Consumer Group運行程序中，如果新創建一個滿足正則匹配條件的Topic，那么Consumer Group就會觸發Rebalance，
3. 訂閱Topic的磁區數發生變更，Kafka當前只能允許增加一個Topic的磁區數，當磁區數增加時，就會觸發訂閱Topic的所有Consumer Group開啟Rebalance，
   Rebalance發生時，Consumer Group下所有的Consumer實體都會協調在一起共同參與，Kafka默認提供了3種Topic磁區的分配策略，每個Consumer實體都能夠得到較為平均的磁區數，

Rebalance的缺點如下：

1. Rebalance程序對Consumer Group消費程序有極大的影響，在Rebalance程序中，所有Consumer實體都會停止消費，等待Rebalance完成，
2. Rebalance程序中所有Consumer實體共同參與，全部重新分配所有磁區，更高效的做法是盡量減少分配方案的變動，
3. Rebalance程序太慢，

• Coordinator
  Coordinator一般指的是運行在每個Broker上的Group Coordinator行程，用于管理Consumer Group中的各個成員，主要用于offset位移管理和Rebalance，一個Coordinator可以同時管理多個消費者組，
  kafka引入Coordinator有其歷史背景，原來consumer資訊依賴于Zookeeper存盤，當代理或消費者發生變化時，引發消費者平衡，此時消費者之間是互不透明的，每個消費者和Zookeeper單獨通信，容易造成羊群效應和腦裂問題，
  為了解決這些問題，kafka引入了Coordinator，服務端引入組協調器（Group Coordinator），消費者端引入消費者協調器（Consumer Coordinator），每個Broker啟動的時候，都會創建Group Coordinator實體，管理部分消費組（集群負載均衡）和組下每個消費者消費的偏移量（offset），每個Consumer實體化時，同時實體化一個Consumer Coordinator物件，負責同一個消費組下各個消費者和服務端組協調器之前的通信，

3.4 資料可靠性

• 當producer向leader發送資料時，可以通過request.required.acks引數來設定資料可靠性的級別

1（默認）： 這意味著producer在ISR中的leader已成功收到資料并得到確認，如果leader宕機了，則會丟失資料，

0： 這意味著producer無需等待來自broker的確認而繼續發送下一批訊息，這種情況下資料傳輸效率最高，但是資料可靠性確是最低的，

-1： producer需要等待ISR中的所有follower都確認接收到資料后才算一次發送完成，可靠性最高，但是這樣也不能保證資料不丟失，比如當ISR中只有leader時（前面ISR那一節講到，ISR中的成員由于某些情況會增加也會減少，最少就只剩一個leader），這樣就變成了acks=1的情況，

如果要提高資料的可靠性，在設定request.required.acks=-1的同時，也要min.insync.replicas這個引數(可以在broker或者topic層面進行設定)的配合，這樣才能發揮最大的功效，min.insync.replicas這個引數設定ISR中的最小副本數是多少，默認值為1，當且僅當request.required.acks引數設定為-1時，此引數才生效，如果ISR中的副本數少于min.insync.replicas配置的數量時，客戶端會回傳例外：org.apache.kafka.common.errors.NotEnoughReplicasExceptoin: Messages are rejected since there are fewer in-sync replicas than required，

• 要保證資料寫入到Kafka是安全的，高可靠的，需要如下的配置：

topic的配置：replication.factor>=3,即副本數至少是3個；2<=min.insync.replicas<=replication.factor

broker的配置：leader的選舉條件unclean.leader.election.enable=false

producer的配置：request.required.acks=-1(all)，producer.type=sync

4 安裝部署及常用命令

4 安裝部署及常用命令

4.1 安裝部署

• 環境描述

zk 3.6.3 172.16.212.21/172.16.212.22/172.16.212.23
java 1.8 172.16.212.21/172.16.212.22/172.16.212.23
kafka 2.8.0 172.16.212.21/172.16.212.22/172.16.212.23

• 安裝配置java環境和zookeeper

詳見 https://blog.csdn.net/qq_35550345/article/details/116208174

• 3個節點：安裝kafka

wget https://mirror-hk.koddos.net/apache/kafka/2.8.0/kafka_2.13-2.8.0.tgz
mkdir -p /opt/work
mkdir -p /opt/data/kafka
tar xvfz kafka_2.13-2.8.0.tgz -C /opt/work
ln -s /opt/work/kafka_2.13-2.8.0 /opt/work/kafka

# 編輯組態檔 server.properties
vim /opt/work/kafka_2.13-2.8.0/conf/server.properties
##Broker的ID，每個broker必須又有唯一的值
broker.id=1
##節點ip和埠
listeners=PLAINTEXT://172.16.212.21:9092 
advertised.listeners=PLAINTEXT://172.16.212.21:9092
num.network.threads=3
num.io.threads=8
num.partitions=1
num.recovery.threads.per.data.dir=3
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/data/kafka
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=172.16.212.21:2181,172.16.212.22:2181,172.16.212.23:2181
zookeeper.connection.timeout.ms=6000
min.insync.replicas = 2
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3
auto.create.topics.enable=false

# 啟動
# /opt/work/kafka_2.13-2.8.0/bin/kafka-server-start.sh --daemon ../config/server.properties

# systemd管理kafka
vim /etc/systemd/system/kafka.service
[Unit]
Description=Apache Kafka server (broker)
After=network.target  zookeeper.service
  
[Service]
Type=simple
User=root
Group=root
Environment=LOG_DIR=/opt/logs/kafka
ExecStart=/opt/work/kafka_2.13-2.8.0/bin/kafka-server-start.sh /opt/work/kafka/config/server.properties
ExecStop=/opt/work/kafka_2.13-2.8.0/bin/kafka-server-stop.sh
Restart=always
  
[Install]
WantedBy=multi-user.target

systemctl daemon-reload
systemctl start kafka
# 報錯了 /opt/work/kafka_2.13-2.8.0/bin/kafka-run-class.sh: line 330: exec: java: not found
# 修復方法 ln -s /usr/local/jdk1.8.0_261/bin/java /usr/bin/java
systemctl enable kafka
systemctl status kafka

• 測驗

# 啟動producer
./kafka-console-producer.sh --broker-list 172.16.212.21:9092 --topic test
# 隨便輸入資訊

# 查看訊息是否到達了broker
./kafka-console-consumer.sh --bootstrap-server 172.16.212.21:9092 --topic test --from-beginning
# 這邊都能看到隨便輸入的資訊

# 查看comsumer有沒有在正常消費到資料
# 消費test 最開始至現在所有的資料
./kafka-console-consumer.sh --bootstrap-server 172.16.212.21:9092 --topic test --from-beginning
# 只消費當前之后所有資料
./kafka-console-consumer.sh --bootstrap-server 172.16.212.21:9092 --topic test

4.2 組態檔詳解

• Broker配置

broker.id=1

Broker的ID，每個broker必須又有唯一的值，

listeners=PLAINTEXT://172.16.3.177:9092

Socket服務器監聽的地址，如果沒有設定，則監聽java.net.InetAddress.getCanonicalHostName()回傳的地址

advertised.listeners=PLAINTEXT://your.host.name:9092

broker通知到producers和consumers的主機地址和埠號，如果未設定，使用listeners的配置，否則，使用java.net.InetAddress#.getCanonicalHostName()回傳的值，如果均無設定，拿到的值為localhot:9092,將造成無法連接其他機器

listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

listener名稱到安全協議，默認所有的listener使用相同的安全協議，PLAINTEXT為明碼文本協議

num.network.threads=3

服務器用來從網路接收請求和發送回應資料到網路的執行緒數

num.io.threads=8

服務器用來處理請求的執行緒數，可能包含磁盤I/O的執行緒，

socket.send.buffer.bytes=102400

服務器發送資料的快取大小，

socket.receive.buffer.bytes=102400

服務器接收資料的快取大小，

socket.request.max.bytes=104857600

socket服務器接收的單個request大小的最大值，

log.dirs=/home/logs/kafka

Kafka存盤log檔案的目錄，多個目錄用逗號分隔，

num.partitions=10

topic的默認分割的磁區個數，多磁區允許消費者并行獲取資料，但這也會造成brokers交叉保存多個檔案，

num.recovery.threads.per.data.dir=2

當Kafka啟動時恢復資料和關閉時保存資料到磁盤時使用的執行緒個數，

log.flush.interval.messages=10000

在持久化到磁盤前message最大接收條數，

log.flush.interval.ms=1000

持久化的最大時間間隔，

log.retention.hours=168

將已保存超過7天的資料洗掉，

log.retention.bytes=1073741824

基于資料大小的log保存策略，當分片的大小超過該值時，就會被洗掉，改功能不依賴于log.retention.hours，

log.segment.bytes=1073741824

單個分片的上限，達到該大小后會生成新的日志分片，

log.retention.check.interval.ms=300000

日志分片的檢測時間間隔，每隔該事件會根據log保留策略決定是否洗掉log分片，

zookeeper.connect=172.16.3.177:12181,172.16.3.178:12181,172.16.3.179:12181

Zookeeper連接字串，是一個使用逗號分隔的host：port字串，

zookeeper.connection.timeout.ms=6000

連接zookeeper的超時時間，

group.initial.rebalance.delay.ms=0

在開發測驗環境下該值設定為0，保證啟動后馬上可以使用，但在生產環境下，默認值3秒更適合，

unclean.leader.election.enable = false

默認值是false, 是否允許ISR集合外的副本參與leader選舉, 如果引數設定為true，就有可能發生資料丟失和資料不一致的情況，Kafka的可靠性就會降低；而如果引數設定為false，Kafka的可用性就會降低

min.insync.replicas = 2

當producer設定request.required.acks為-1時，min.insync.replicas指定replicas的最小數目（必須確認每一個repica的寫資料都是成功的），如果這個數目沒有達到，producer會產生例外，

• Topic配置

cleanup.policy
Default(默認值)：delete

server.properties：log.cleanup.policy
說明(解釋)：日志清理策略選擇有：delete和compact主要針對過期資料的處理，或是日志檔案達到限制的額度，會被 topic創建時的指定引數覆寫

delete.retention.ms
Default(默認值)：86400000 (24 hours)

server.properties：log.cleaner.delete.retention.ms
說明(解釋)： 對于壓縮的日志保留的最長時間，也是客戶端消費訊息的最長時間，同log.retention.minutes的區別在于一個控制未壓縮資料，一個控制壓縮后的資料，會被topic創建時的指定引數覆寫

flush.messages
Default(默認值)：None

server.properties：log.flush.interval.messages
說明(解釋)： log檔案”sync”到磁盤之前累積的訊息條數,因為磁盤IO操作是一個慢操作,但又是一個”資料可靠性"的必要手段,所以此引數的設定,需要在"資料可靠性"與"性能"之間做必要的權衡.如果此值過大,將會導致每次"fsync"的時間較長(IO阻塞),如果此值過小,將會導致"fsync"的次數較多,這也意味著整體的client請求有一定的延遲.物理server故障,將會導致沒有fsync的訊息丟失.

flush.ms
Default(默認值)：None

server.properties：log.flush.interval.ms
說明(解釋)：僅僅通過interval來控制訊息的磁盤寫入時機,是不足的.此引數用于控制"fsync"的時間間隔,如果訊息量始終沒有達到閥值,但是離上一次磁盤同步的時間間隔達到閥值,也將觸發.

index.interval.bytes
Default(默認值)：4096

server.properties：log.index.interval.bytes
說明(解釋)：當執行一個fetch操作后，需要一定的空間來掃描最近的offset大小，設定越大，代表掃描速度越快，但是也更耗記憶體，一般情況下不需要搭理這個引數

message.max.bytes
Default(默認值)：1,000,000

server.properties： 表示訊息的最大大小，單位是位元組 說明(解釋)：當執行一個fetch操作后，需要一定的空間來掃描最近的offset大小，設定越大，代表掃描速度越快，但是也更好記憶體，一般情況下不需要搭理這個引數

min.cleanable.dirty.ratio
Default(默認值)：0.5

server.properties：log.cleaner.min.cleanable.ratio
說明(解釋)：日志清理的頻率控制，越大意味著更高效的清理，同時會存在一些空間上的浪費，會被topic創建時的指定引數覆寫

retention.bytes
Default(默認值)：None

server.properties：log.retention.bytes
說明(解釋)： topic每個磁區的最大檔案大小，一個topic的大小限制 = 磁區數*log.retention.bytes，-1沒有大小限log.retention.bytes和log.retention.minutes任意一個達到要求，都會執行洗掉，會被topic創建時的指定引數覆寫

retention.ms
Default(默認值)：None

server.properties：log.retention.minutes
說明(解釋)： 資料存盤的最大時間超過這個時間會根據log.cleanup.policy設定的策略處理資料，也就是消費端能夠多久去消費資料 log.retention.bytes和log.retention.minutes達到要求，都會執行洗掉，會被topic創建時的指定引數覆寫

segment.bytes
Default(默認值)：1 GB

server.properties：log.segment.bytes
說明(解釋)： topic的磁區是以一堆segment檔案存盤的，這個控制每個segment的大小，會被topic創建時的指定引數覆寫

segment.index.bytes
Default(默認值)：10 MB

server.properties：log.index.size.max.bytes
說明(解釋)：對于segment日志的索引檔案大小限制，會被topic創建時的指定引數覆寫

log.roll.hours
Default(默認值)：7 days

server.properties：log.index.size.max.bytes
說明(解釋)： 這個引數會在日志segment沒有達到log.segment.bytes設定的大小，也會強制新建一個segment會被 topic創建時的指定引數覆寫

4.3 常用命令

cd /opt/work/kafka/bin

# 創建topic
./kafka-topics.sh --bootstrap-server 172.16.212.21:9092 --create --topic test --partitions 2 --replication-factor 3

# 查看topic資訊
./kafka-topics.sh --bootstrap-server 172.16.212.21:9092 --describe --topic test

# 增加指定topic的磁區數量
./kafka-topics.sh --bootstrap-server 172.16.212.21:9092 --alter --topic test --partitions 3

# 洗掉topic
./kafka-topics.sh --bootstrap-server 172.16.212.21:9092 --delete --topic test

# 查看訊息堆積情況
./kafka-consumer-groups.sh --bootstrap-server 172.16.212.21:9092 --describe --group test

# 查看正在運行的消費組
./kafka-consumer-groups.sh --bootstrap-server 172.16.212.21:9092 --list

5 與其他訊息系統對比