關注公衆号：<code>大資料技術派</code>，回複: <code>資料</code>，領取<code>1024G</code>資料。
高吞吐量、低延遲：kafka每秒可以處理幾十萬條消息，它的延遲最低隻有幾毫秒，每個topic可以分多個partition, consumer group 對partition進行consume操作。
可擴充性：kafka叢集支援熱擴充
持久性、可靠性：消息被持久化到本地磁盤，并且支援資料備份防止資料丢失
容錯性：允許叢集中節點失敗（若副本數量為n,則允許n-1個節點失敗）
高并發：支援數千個用戶端同時讀寫
日志收集：一個公司可以用Kafka可以收集各種服務的log，通過kafka以統一接口服務的方式開放給各種consumer，例如hadoop、HBase、Solr等。
消息系統：解耦和生産者和消費者、緩存消息等。
使用者活動跟蹤：Kafka經常被用來記錄web使用者或者app使用者的各種活動，如浏覽網頁、搜尋、點選等活動，這些活動資訊被各個伺服器釋出到kafka的topic中，然後訂閱者通過訂閱這些topic來做實時的監控分析，或者裝載到hadoop、資料倉庫中做離線分析和挖掘。
營運名額：Kafka也經常用來記錄營運監控資料。包括收集各種分布式應用的資料，生産各種操作的集中回報，比如報警和報告。
流式處理：比如spark streaming和 Flink
簡單架構如下：
詳細架構如下：
Kafka 架構分為以下幾個部分：
Producer：消息生産者，就是向 kafka broker 發消息的用戶端。
Consumer：消息消費者，向 kafka broker 取消息的用戶端。
Topic：可以了解為一個隊列，一個 Topic 又分為一個或多個分區。
Consumer Group：這是 kafka 用來實作一個 topic 消息的廣播（發給所有的 consumer）和單點傳播（發給任意一個 consumer）的手段。一個 topic 可以有多個 Consumer Group。
Broker：一台 kafka 伺服器就是一個 broker。一個叢集由多個 broker 組成。一個 broker 可以容納多個 topic。
Partition：為了實作擴充性，一個非常大的 topic 可以分布到多個 broker上，每個 partition 是一個有序的隊列。partition 中的每條消息都會被配置設定一個有序的id（offset）。将消息發給 consumer，kafka 隻保證按一個 partition 中的消息的順序，不保證一個 topic 的整體（多個 partition 間）的順序。
Offset：kafka 的存儲檔案都是按照 offset.kafka 來命名，用 offset 做名字的好處是友善查找。例如你想找位于 2049 的位置，隻要找到 2048.kafka 的檔案即可。當然 the first offset 就是 00000000000.kafka。
分區對于 Kafka 叢集的好處是：實作負載均衡。分區對于消費者來說，可以提高并發度，提高效率。
kafka 中的每個 partition 中的消息在寫入時都是有序的，而且單獨一個 partition 隻能由一個消費者去消費，可以在裡面保證消息的順序性。但是分區之間的消息是不保證有序的。
可回答：
Kafka 在什麼情況下會出現消息丢失？
1）資料可靠性（可回答 怎麼盡可能保證 Kafka 的可靠性？）
Kafka 作為一個商業級消息中間件，消息可靠性的重要性可想而知。本文從 Producter 往 Broker 發送消息、Topic 分區副本以及 Leader 選舉幾個角度介紹資料的可靠性。
Topic分區副本
在 Kafka 0.8.0 之前，Kafka 是沒有副本的概念的，那時候人們隻會用 Kafka 存儲一些不重要的資料，因為沒有副本，資料很可能會丢失。但是随着業務的發展，支援副本的功能越來越強烈，是以為了保證資料的可靠性，Kafka 從 0.8.0 版本開始引入了分區副本（詳情請參見 KAFKA-50）。也就是說每個分區可以人為的配置幾個副本（比如建立主題的時候指定 replication-factor，也可以在 Broker 級别進行配置 default.replication.factor），一般會設定為3。
Kafka 可以保證單個分區裡的事件是有序的，分區可以線上（可用），也可以離線（不可用）。在衆多的分區副本裡面有一個副本是 Leader，其餘的副本是 follower，所有的讀寫操作都是經過 Leader 進行的，同時 follower 會定期地去 leader 上的複制資料。當 Leader 挂了的時候，其中一個 follower 會重新成為新的 Leader。通過分區副本，引入了資料備援，同時也提供了 Kafka 的資料可靠性。
Kafka 的分區多副本架構是 Kafka 可靠性保證的核心，把消息寫入多個副本可以使 Kafka 在發生崩潰時仍能保證消息的持久性。
Producer 往 Broker 發送消息
如果我們要往 Kafka 對應的主題發送消息，我們需要通過 Producer 完成。前面我們講過 Kafka 主題對應了多個分區，每個分區下面又對應了多個副本；為了讓使用者設定資料可靠性， Kafka 在 Producer 裡面提供了消息确認機制。也就是說我們可以通過配置來決定消息發送到對應分區的幾個副本才算消息發送成功。可以在定義 Producer 時通過 acks 參數指定（在 0.8.2.X 版本之前是通過 request.required.acks 參數設定的）。
這個參數支援以下三種值：
acks = 0：意味着如果生産者能夠通過網絡把消息發送出去，那麼就認為消息已成功寫入Kafka。在這種情況下還是有可能發生錯誤，比如發送的對象無能被序列化或者網卡發生故障，但如果是分區離線或整個叢集長時間不可用，那就不會收到任何錯誤。在 acks=0 模式下的運作速度是非常快的（這就是為什麼很多基準測試都是基于這個模式），你可以得到驚人的吞吐量和帶寬使用率，不過如果選擇了這種模式， 一定會丢失一些消息。
acks = 1：意味若 Leader 在收到消息并把它寫入到分區資料檔案（不一定同步到磁盤上）時會傳回确認或錯誤響應。在這個模式下，如果發生正常的 Leader 選舉，生産者會在選舉時收到一個 LeaderNotAvailableException 異常，如果生産者能恰當地處理這個錯誤，它會重試發送悄息，最終消息會安全到達新的 Leader 那裡。不過在這個模式下仍然有可能丢失資料，比如消息已經成功寫入 Leader，但在消息被複制到 follower 副本之前 Leader發生崩潰。
acks = all（這個和 request.required.acks = -1 含義一樣）：意味着 Leader 在傳回确認或錯誤響應之前，會等待所有同步副本都收到悄息。如果和 min.insync.replicas 參數結合起來，就可以決定在傳回确認前至少有多少個副本能夠收到悄息，生産者會一直重試直到消息被成功送出。不過這也是最慢的做法，因為生産者在繼續發送其他消息之前需要等待所有副本都收到目前的消息。
根據實際的應用場景，我們設定不同的 acks，以此保證資料的可靠性。
另外，Producer 發送消息還可以選擇同步（預設，通過 producer.type=sync 配置） 或者異步（producer.type=async）模式。如果設定成異步，雖然會極大的提高消息發送的性能，但是這樣會增加丢失資料的風險。如果需要確定消息的可靠性，必須将 producer.type 設定為 sync。
Leader 選舉
在介紹 Leader 選舉之前，讓我們先來了解一下 ISR（in-sync replicas）清單。每個分區的 leader 會維護一個 ISR 清單，ISR 清單裡面就是 follower 副本的 Borker 編号，隻有跟得上 Leader 的 follower 副本才能加入到 ISR 裡面，這個是通過 replica.lag.time.max.ms 參數配置的。隻有 ISR 裡的成員才有被選為 leader 的可能。
2）資料一緻性（可回答 Kafka資料一緻性原理？）
這裡介紹的資料一緻性主要是說不論是老的 Leader 還是新選舉的 Leader，Consumer 都能讀到一樣的資料。那麼 Kafka 是如何實作的呢？
假設分區的副本為3，其中副本0是 Leader，副本1和副本2是 follower，并且在 ISR 清單裡面。雖然副本0已經寫入了 Message4，但是 Consumer 隻能讀取到 Message2。因為所有的 ISR 都同步了 Message2，隻有 High Water Mark 以上的消息才支援 Consumer 讀取，而 High Water Mark 取決于 ISR 清單裡面偏移量最小的分區，對應于上圖的副本2，這個很類似于木桶原理。
這樣做的原因是還沒有被足夠多副本複制的消息被認為是“不安全”的，如果 Leader 發生崩潰，另一個副本成為新 Leader，那麼這些消息很可能丢失了。如果我們允許消費者讀取這些消息，可能就會破壞一緻性。試想，一個消費者從目前 Leader（副本0） 讀取并處理了 Message4，這個時候 Leader 挂掉了，選舉了副本1為新的 Leader，這時候另一個消費者再去從新的 Leader 讀取消息，發現這個消息其實并不存在，這就導緻了資料不一緻性問題。
當然，引入了 High Water Mark 機制，會導緻 Broker 間的消息複制因為某些原因變慢，那麼消息到達消費者的時間也會随之變長（因為我們會先等待消息複制完畢）。延遲時間可以通過參數 replica.lag.time.max.ms 參數配置，它指定了副本在複制消息時可被允許的最大延遲時間。
ISR：In-Sync Replicas 副本同步隊列
OSR：Out-of-Sync Replicas
AR：Assigned Replicas 所有副本
ISR是由leader維護，follower從leader同步資料有一些延遲（具體可以參見 圖文了解 Kafka 的副本複制機制），超過相應的門檻值會把 follower 剔除出 ISR, 存入OSR（Out-of-Sync Replicas ）清單，新加入的follower也會先存放在OSR中。AR=ISR+OSR。
LEO：是 LogEndOffset 的簡稱，代表目前日志檔案中下一條
HW：水位或水印（watermark）一詞，也可稱為高水位(high watermark)，通常被用在流式處理領域（比如Apache Flink、Apache Spark等），以表征元素或事件在基于時間層面上的進度。在Kafka中，水位的概念反而與時間無關，而是與位置資訊相關。嚴格來說，它表示的就是位置資訊，即位移（offset）。取 partition 對應的 ISR中 最小的 LEO 作為 HW，consumer 最多隻能消費到 HW 所在的位置上一條資訊。
LSO：是 LastStableOffset 的簡稱，對未完成的事務而言，LSO 的值等于事務中第一條消息的位置(firstUnstableOffset)，對已完成的事務而言，它的值同 HW 相同
LW：Low Watermark 低水位, 代表 AR 集合中最小的 logStartOffset 值。
資料傳輸的事務定義通常有以下三種級别：
最多一次：消息不會被重複發送，最多被傳輸一次，但也有可能一次不傳輸
最少一次：消息不會被漏發送，最少被傳輸一次，但也有可能被重複傳輸
精确的一次（Exactly once）：不會漏傳輸也不會重複傳輸，每個消息都傳輸被接收
Kafa consumer消費消息時，向broker發出fetch請求去消費特定分區的消息，consumer指定消息在日志中的偏移量（offset），就可以消費從這個位置開始的消息，customer擁有了offset的控制權，可以向後復原去重新消費之前的消息，這是很有意義的。
Kafka最初考慮的問題是，customer應該從brokes拉取消息還是brokers将消息推送到consumer，也就是pull還push。在這方面，Kafka遵循了一種大部分消息系統共同的傳統的設計：producer将消息推送到broker，consumer從broker拉取消息。
一些消息系統比如Scribe和Apache Flume采用了push模式，将消息推送到下遊的consumer。這樣做有好處也有壞處：由broker決定消息推送的速率，對于不同消費速率的consumer就不太好處理了。消息系統都緻力于讓consumer以最大的速率最快速的消費消息，但不幸的是，push模式下，當broker推送的速率遠大于consumer消費的速率時，consumer恐怕就要崩潰了。最終Kafka還是選取了傳統的pull模式。
Pull模式的另外一個好處是consumer可以自主決定是否批量的從broker拉取資料。Push模式必須在不知道下遊consumer消費能力和消費政策的情況下決定是立即推送每條消息還是緩存之後批量推送。如果為了避免consumer崩潰而采用較低的推送速率，将可能導緻一次隻推送較少的消息而造成浪費。Pull模式下，consumer就可以根據自己的消費能力去決定這些政策。 Pull有個缺點是，如果broker沒有可供消費的消息，将導緻consumer不斷在循環中輪詢，直到新消息到t達。為了避免這點，Kafka有個參數可以讓consumer阻塞知道新消息到達(當然也可以阻塞知道消息的數量達到某個特定的量這樣就可以批量發送）
1）Kafka把topic中一個parition大檔案分成多個小檔案段，通過多個小檔案段，就容易定期清除或删除已經消費完檔案，減少磁盤占用。
2）通過索引資訊可以快速定位message和确定response的最大大小。
3）通過index中繼資料全部映射到memory，可以避免segment file的IO磁盤操作。
4）通過索引檔案稀疏存儲，可以大幅降低index檔案中繼資料占用空間大小。
1）副本因子不能大于 Broker 的個數；
2）第一個分區（編号為0）的第一個副本放置位置是随機從 brokerList 選擇的；
3）其他分區的第一個副本放置位置相對于第0個分區依次往後移。也就是如果我們有5個 Broker，5個分區，假設第一個分區放在第四個 Broker 上，那麼第二個分區将會放在第五個 Broker 上；第三個分區将會放在第一個 Broker 上；第四個分區将會放在第二個 Broker 上，依次類推；
4）剩餘的副本相對于第一個副本放置位置其實是由 nextReplicaShift 決定的，而這個數也是随機産生的；
我們知道，在啟動 Kafka 叢集之前，我們需要配置好 log.dirs 參數，其值是 Kafka 資料的存放目錄，這個參數可以配置多個目錄，目錄之間使用逗号分隔，通常這些目錄是分布在不同的磁盤上用于提高讀寫性能。當然我們也可以配置 log.dir 參數，含義一樣。隻需要設定其中一個即可。
如果 log.dirs 參數隻配置了一個目錄，那麼配置設定到各個 Broker 上的分區肯定隻能在這個目錄下建立檔案夾用于存放資料。
但是如果 log.dirs 參數配置了多個目錄，那麼 Kafka 會在哪個檔案夾中建立分區目錄呢？答案是：Kafka 會在含有分區目錄最少的檔案夾中建立新的分區目錄，分區目錄名為 Topic名+分區ID。注意，是分區檔案夾總數最少的目錄，而不是磁盤使用量最少的目錄！也就是說，如果你給 log.dirs 參數新增了一個新的磁盤，新的分區目錄肯定是先在這個新的磁盤上建立直到這個新的磁盤目錄擁有的分區目錄不是最少為止。
在Kafka中，當有新消費者加入或者訂閱的topic數發生變化時，會觸發Rebalance(再均衡：在同一個消費者組當中，分區的所有權從一個消費者轉移到另外一個消費者)機制，Rebalance顧名思義就是重新均衡消費者消費。Rebalance的過程如下：
第一步：所有成員都向coordinator發送請求，請求入組。一旦所有成員都發送了請求，coordinator會從中選擇一個consumer擔任leader的角色，并把組成員資訊以及訂閱資訊發給leader。
第二步：leader開始配置設定消費方案，指明具體哪個consumer負責消費哪些topic的哪些partition。一旦完成配置設定，leader會将這個方案發給coordinator。coordinator接收到配置設定方案之後會把方案發給各個consumer，這樣組内的所有成員就都知道自己應該消費哪些分區了。
是以對于Rebalance來說，Coordinator起着至關重要的作用。
在 Kafka 内部存在兩種預設的分區配置設定政策：Range 和 RoundRobin。當以下事件發生時，Kafka 将會進行一次分區配置設定：
1）同一個 Consumer Group 内新增消費者
2）消費者離開目前所屬的Consumer Group，包括shuts down 或 crashes
3）訂閱的主題新增分區
将分區的所有權從一個消費者移到另一個消費者稱為重新平衡（rebalance），如何rebalance就涉及到下面提到的分區配置設定政策。下面我們将詳細介紹 Kafka 内置的兩種分區配置設定政策。本文假設我們有個名為 T1 的主題，其包含了10個分區，然後我們有兩個消費者（C1，C2）來消費這10個分區裡面的資料，而且 C1 的 num.streams = 1，C2 的 num.streams = 2。
Range strategy
Range政策是對每個主題而言的，首先對同一個主題裡面的分區按照序号進行排序，并對消費者按照字母順序進行排序。在我們的例子裡面，排完序的分區将會是0, 1, 2, 3, 4, 5, 6, 7, 8, 9；消費者線程排完序将會是C1-0, C2-0, C2-1。然後将partitions的個數除于消費者線程的總數來決定每個消費者線程消費幾個分區。如果除不盡，那麼前面幾個消費者線程将會多消費一個分區。
在我們的例子裡面，我們有10個分區，3個消費者線程，10 / 3 = 3，而且除不盡，那麼消費者線程 C1-0 将會多消費一個分區，是以最後分區配置設定的結果看起來是這樣的：
C1-0 将消費 0, 1, 2, 3 分區
C2-0 将消費 4, 5, 6 分區
C2-1 将消費 7, 8, 9 分區
假如我們有11個分區，那麼最後分區配置設定的結果看起來是這樣的：
C2-0 将消費 4, 5, 6, 7 分區
C2-1 将消費 8, 9, 10 分區
假如我們有2個主題(T1和T2)，分别有10個分區，那麼最後分區配置設定的結果看起來是這樣的：
C1-0 将消費 T1主題的 0, 1, 2, 3 分區以及 T2主題的 0, 1, 2, 3分區
C2-0 将消費 T1主題的 4, 5, 6 分區以及 T2主題的 4, 5, 6分區
C2-1 将消費 T1主題的 7, 8, 9 分區以及 T2主題的 7, 8, 9分區
可以看出，C1-0 消費者線程比其他消費者線程多消費了2個分區，這就是Range strategy的一個很明顯的弊端。
RoundRobin strategy
使用RoundRobin政策有兩個前提條件必須滿足：
同一個Consumer Group裡面的所有消費者的num.streams必須相等；
每個消費者訂閱的主題必須相同。
是以這裡假設前面提到的2個消費者的num.streams = 2。RoundRobin政策的工作原理：将所有主題的分區組成 TopicAndPartition 清單，然後對 TopicAndPartition 清單按照 hashCode 進行排序，這裡文字可能說不清，看下面的代碼應該會明白：
最後按照round-robin風格将分區分别配置設定給不同的消費者線程。
在我們的例子裡面，假如按照 hashCode 排序完的topic-partitions組依次為T1-5, T1-3, T1-0, T1-8, T1-2, T1-1, T1-4, T1-7, T1-6, T1-9，我們的消費者線程排序為C1-0, C1-1, C2-0, C2-1，最後分區配置設定的結果為：
C1-0 将消費 T1-5, T1-2, T1-6 分區；
C1-1 将消費 T1-3, T1-1, T1-9 分區；
C2-0 将消費 T1-0, T1-4 分區；
C2-1 将消費 T1-8, T1-7 分區。
多個主題的分區配置設定和單個主題類似。
Kafka是分布式消息系統，需要處理海量的消息，Kafka的設計是把所有的消息都寫入速度低容量大的硬碟，以此來換取更強的存儲能力，但實際上，使用硬碟并沒有帶來過多的性能損失。kafka主要使用了以下幾個方式實作了超高的吞吐率：
1）順序讀寫
2）零拷貝
3）檔案分段
4）批量發送
5）資料壓縮
1）由于是批量發送，資料并非真正的實時；
2）對于mqtt協定不支援；
3）不支援物聯網傳感資料直接接入；
4）僅支援統一分區内消息有序，無法實作全局消息有序；
5）監控不完善，需要安裝插件；
6）依賴zookeeper進行中繼資料管理。
舊的 Kafka 消費者 API 主要包括：SimpleConsumer（簡單消費者） 和 ZookeeperConsumerConnectir（進階消費者）。SimpleConsumer 名字看起來是簡單消費者，但是其實用起來很不簡單，可以使用它從特定的分區和偏移量開始讀取消息。進階消費者和現在新的消費者有點像，有消費者群組，有分區再均衡，不過它使用 ZK 來管理消費者群組，并不具備偏移量和再均衡的可操控性。
現在的消費者同時支援以上兩種行為，是以為啥還用舊消費者 API 呢？
我們可以使用 bin/kafka-topics.sh 指令對 Kafka 增加 Kafka 的分區資料，但是 Kafka 不支援減少分區數。 Kafka 分區資料不支援減少是由很多原因的，比如減少的分區其資料放到哪裡去？是删除，還是保留？删除的話，那麼這些沒消費的消息不就丢了。如果保留這些消息如何放到其他分區裡面？追加到其他分區後面的話那麼就破壞了 Kafka 單個分區的有序性。如果要保證删除分區資料插入到其他分區保證有序性，那麼實作起來邏輯就會非常複雜。
猜你喜歡
Hive計算最大連續登陸天數
Hadoop 資料遷移用法詳解
Hbase修複工具Hbck
數倉模組化分層理論
一文搞懂Hive的資料存儲與壓縮
大資料元件重點學習這幾個