Kafka之工作流程分析

kafka核心組成

一、Kafka生產過程分析

1.1 寫入方式

　　producer採用推（push）模式將訊息釋出到broker，每條訊息都被追加（append）到分割槽（patition）中，屬於順序寫磁碟（順序寫磁碟效率比隨機寫記憶體要高，保障kafka吞吐率）。

1.2 分割槽(Partition)

　　訊息傳送時都被髮送到一個topic，其本質就是一個目錄，而topic是由一些Partition Logs(分割槽日誌)組成，其組織結構如下圖所示：　　

　　我們可以看到，每個Partition中的訊息都是有序的，生產的訊息被不斷追加到Partition log上，其中的每一個訊息都被賦予了一個唯一的offset值

（1）分割槽的原因
　　1）方便在叢集中擴充套件，每個Partition可以通過調整以適應它所在的機器，而一個topic又可以有多個Partition組成，因此整個叢集就可以適應任意大小的資料了；
　　2）可以提高併發，因為可以以Partition為單位讀寫了。
（2）分割槽的原則
　　1）指定了patition，則直接使用；
　　2）未指定patition但指定key，通過對key的value進行hash出一個patition；
　　3）patition和key都未指定，使用輪詢選出一個patition。

DefaultPartitioner類
public int partition(String topic, Object key, byte
 
[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        if (keyBytes == null) {
            int nextValue = nextValue(topic);
            List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
            
 
if (availablePartitions.size() > 0) {
                int part = Utils.toPositive(nextValue) % availablePartitions.size();
                return availablePartitions.get(part).partition();
            } else {
                // no partitions are available, give a non-available partition
                return Utils.toPositive(nextValue) % numPartitions;
            }
        } else {
            // hash the keyBytes to choose a partition
            return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
        }
    }

1.3副本（Replication）

　　同一個partition可能會有多個replication（對應 server.properties 配置中的 default.replication.factor=N）。沒有replication的情況下，一旦broker 宕機，其上所有 patition 的資料都不可被消費，同時producer也不能再將資料存於其上的patition。引入replication之後，同一個partition可能會有多個replication，而這時需要在這些replication之間選出一個leader，producer和consumer只與這個leader互動，其它replication作為follower從leader 中複製資料。

1.4寫入流程

producer寫入訊息流程如下：

1、Broker：安裝Kafka服務的那臺叢集就是一個broker（broker的id要全域性唯一）
2、Producer：訊息的生產者，負責將資料寫入到broker中（push）
3、Consumer：訊息的消費者，負責從kafka中讀取資料（pull），老版本的消費者需要依賴zk，新版本的不需要
4、Topic:主題，相當於是資料的一個分類，不同topic存放不同的資料
5、replication：副本，資料儲存多少份（保證資料不丟）
6、partition：分割槽，是一個物理分割槽，一個分割槽就是一個檔案，一個topic可以有一到多個分割槽，每一個分割槽都有自己的副本。
7、Consumer Group：消費者組，一個topic可以有多個消費者同時消費，多個消費者如果在一個消費者組中，那麼他們不能重複消費資料

二、Broker 儲存訊息

2.1 儲存方式

　　物理上把topic分成一個或多個patition（對應server.properties 中的num.partitions=3配置），每個patition物理上對應一個資料夾（該資料夾儲存該patition的所有訊息和索引檔案），如下：

[hadoop1 logs]$ ll
drwxrwxr-x. 2 hadoop hadoop  4096 8月   6 14:37 first-0
drwxrwxr-x. 2 hadoop hadoop  4096 8月   6 14:35 first-1
drwxrwxr-x. 2 hadoop hadoop  4096 8月   6 14:37 first-2
[hadoop1 logs]$ cd first-0
[hadoop1 first-0]$ ll
-rw-rw-r--. 1 hadoop hadoop 10485760 8月   6 14:33 00000000000000000000.index
-rw-rw-r--. 1 hadoop hadoop      219 8月   6 15:07 00000000000000000000.log
-rw-rw-r--. 1 hadoop hadoop 10485756 8月   6 14:33 00000000000000000000.timeindex
-rw-rw-r--. 1 hadoop hadoop        8 8月   6 14:37 leader-epoch-checkpoint

2.2 儲存策略

無論訊息是否被消費，kafka都會保留所有訊息。有兩種策略可以刪除舊資料：
　　1）基於時間：log.retention.hours=168
　　2）基於大小：log.retention.bytes=1073741824
需要注意的是，因為Kafka讀取特定訊息的時間複雜度為O(1)，即與檔案大小無關，所以這裡刪除過期檔案與提高 Kafka 效能無關。

2.3 Zookeeper儲存結構

注意：producer不在zk中註冊，消費者在zk中註冊。

三、Kafka消費過程分析

　　kafka提供了兩套consumer API：高階Consumer API和低階Consumer API。

3.1高階API

1）高階API優點
　　高階API 寫起來簡單
　　　　不需要自行去管理offset，系統通過zookeeper自行管理。
　　　　不需要管理分割槽，副本等情況，系統自動管理。
　　消費者斷線會自動根據上一次記錄在zookeeper中的offset去接著獲取資料（預設設定1分鐘更新一下zookeeper中存的offset）
　　可以使用group來區分對同一個topic 的不同程式訪問分離開來（不同的group記錄不同的offset，這樣不同程式讀取同一個topic才不會因為offset互相影響）
2）高階API缺點
　　不能自行控制offset（對於某些特殊需求來說）
　　不能細化控制如分割槽、副本、zk等

3.2低階API

1）低階 API 優點
　　能夠讓開發者自己控制offset，想從哪裡讀取就從哪裡讀取。
　　自行控制連線分割槽，對分割槽自定義進行負載均衡
　　對zookeeper的依賴性降低（如：offset不一定非要靠zk儲存，自行儲存offset即可，比如存在檔案或者記憶體中）
2）低階API缺點
　　太過複雜，需要自行控制offset，連線哪個分割槽，找到分割槽leader 等。

3.3消費者組

　　消費者是以consumer group消費者組的方式工作，由一個或者多個消費者組成一個組，共同消費一個topic。每個分割槽在同一時間只能由group中的一個消費者讀取，但是多個group可以同時消費這個partition。在圖中，有一個由三個消費者組成的group，有一個消費者讀取主題中的兩個分割槽，另外兩個分別讀取一個分割槽。某個消費者讀取某個分割槽，也可以叫做某個消費者是某個分割槽的擁有者。
　　在這種情況下，消費者可以通過水平擴充套件的方式同時讀取大量的訊息。另外，如果一個消費者失敗了，那麼其他的group成員會自動負載均衡讀取之前失敗的消費者讀取的分割槽。

3.4 消費方式

　　consumer採用pull（拉）模式從broker中讀取資料。

　　push（推）模式很難適應消費速率不同的消費者，因為訊息傳送速率是由broker決定的。它的目標是儘可能以最快速度傳遞訊息，但是這樣很容易造成consumer來不及處理訊息，典型的表現就是拒絕服務以及網路擁塞。而pull模式則可以根據consumer的消費能力以適當的速率消費訊息。
　　對於Kafka而言，pull模式更合適，它可簡化broker的設計，consumer可自主控制消費訊息的速率，同時consumer可以自己控制消費方式——即可批量消費也可逐條消費，同時還能選擇不同的提交方式從而實現不同的傳輸語義。
pull模式不足之處是，如果kafka沒有資料，消費者可能會陷入迴圈中，一直等待資料到達。為了避免這種情況，我們在我們的拉請求中有引數，允許消費者請求在等待資料到達的“長輪詢”中進行阻塞（並且可選地等待到給定的位元組數，以確保大的傳輸大小）。

3.5 消費者組案例

　　1）需求：測試同一個消費者組中的消費者，同一時刻只能有一個消費者消費。
　　2）案例實操

　　（1）在hadoop1、hadoop2上修改/home/bigdata/kafka/config/consumer.properties配置檔案中的group.id屬性為任意組名。

[hadoop2 config]$ vi consumer.properties
group.id=hadoop1

　　（2）在hadoop1、hadoop2上分別啟動消費者

[hadoop1 kafka]$ bin/kafka-console-consumer.sh \
--zookeeper hadoop1:2181 --topic first --consumer.config config/consumer.properties
[hadoop2 kafka]$ bin/kafka-console-consumer.sh --zookeeper hadoop1:2181 --topic first --consumer.config config/consumer.properties

　　（3）在hadoop104上啟動生產者

[hadoop1 kafka]$ bin/kafka-console-producer.sh \
--broker-list hadoop1:9092 --topic first
>hello world

　　（4）檢視hadoop1和hadoop2的接收者

　　　　同一時刻只有一個消費者接收到訊息。