Kafka 生產過程分析

傳送訊息的主要步驟:

我們從建立乙個 producerrecord 物件開始， producerrecord 物件需要包含目標主題和要傳送的內容。

我們還可以指定鍵或分割槽。在傳送 producerrecord 物件時，生產者要先把鍵和值物件序列化成位元組陣列，這樣它們才能夠在網路上傳輸。

然後資料被傳給分割槽器。如果之前在 producerrecord 物件裡指定了分割槽，那麼分割槽器就不會再做任何事情，直接把指定的分割槽返回。如果沒有指定分割槽，那麼分割槽器會根據 producerrecord 物件的鍵來選擇乙個分割槽。

選好分割槽以後，生產者就知道該往哪個主題和分割槽傳送這條記錄了。緊接著，這條記錄被新增到乙個記錄批次裡，這個批次裡的所有訊息會被傳送到相同的主題和分割槽 broker 上。

伺服器在收到這些訊息時會返回乙個ack響應。如果訊息成功寫入 kafka，就返回乙個recordmetadata 物件，它包含了主題和分割槽資訊，以及記錄在分割槽裡的偏移量。如果寫入失敗，則會返回乙個錯誤。生產者在收到錯誤之後會嘗試重新傳送訊息，幾次之後如果還是失敗，就返回錯誤資訊。

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

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

方便在集群中擴充套件，每個 partition 可以通過調整以適應它所在的機器，而乙個topic 又可以有多個 partition 組成，因此整個集群就可以適應任意大小的資料了；

可以提高併發，因為可以以 partition 為單位讀寫了。

指定了 patition，則直接使用；

未指定 patition 但指定 key，通過對 key 的 value 進行 hash 出乙個patition；

patition 和 key 都未指定，使用輪詢選出乙個 patition。

kafakproducer

private int partition(producerrecordrecord, byte serializedkey, byte serializedvalue, cluster cluster)

org.apache.kafka.clients.producer.internals.defaultpartitioner類:

/**
* compute the partition for the given record.
* * @param topic the topic name
* @param key the key to partition on (or null if no key)
* @param keybytes serialized key to partition on (or null if no key)
* @param value the value to partition on or null
* @param valuebytes serialized value to partition on or null
* @param cluster the current cluster metadata
*/public int partition(string topic, object key, byte keybytes, object value, byte valuebytes, cluster cluster)  else 
} else 
}

同乙個 partition 可能會有多個 replication（對應 server.properties 配置中的 default.replication.factor=n）。

沒有 replication 的情況下，一旦 broker 宕機，其上所有 patition 的資料都不可被消費，同時 producer 也不能再將資料存於其上的 patition。

引入 replication 之後，同乙個 partition 可能會有多個 replication，而這時需要在這些 replication 之間選出乙個leader，producer 和 consumer 只與這個leade r互動，其它 replication 作為 follower 從 leader 中複製資料。

一段 leader 掛了, 則會從其他的 follower 中選舉出新的 leader, 新的 leader 就會承擔起讀寫的職責.

複本數不能大於 broker 的數量

producer 先從 zookeeper 的 "/brokers/.../state" 節點找到該partition 的 leader

producer 將訊息傳送給該 leader

leader 將訊息寫入本地 log

followers 從 leader pull 訊息，寫入本地 log 後向 leader 傳送 ack

leader 收到所有 isr 中的 replication 的ack後，增加 hw（high watermark，最後commit 的offset）並向 producer 傳送 ack

ack 是訊息確認模式.

主要有 3 個值:

0意味著如果生產者能夠通過網路把訊息傳送出去，那麼就認為訊息已成功寫入 kafka。在這種情況下還是有可能發生錯誤，比如傳送的物件無法被序列化或者網絡卡發生故障，如果是分割槽離線或整個集群長時間不可用，那就不會收到任何錯誤.

在acks=0模式下的執行速度是非常快的. 你可以得到驚人的吞吐量和頻寬利用率,不過如果選擇了這種模式，一定會丟失一些訊息。

1意味若首領在收到訊息並把它寫入到分割槽資料檔案時會返回確認或錯誤響應. 不過在這個模式下仍然有可能丟失資料，比如訊息已經成功寫入首領，但在訊息被複製到跟隨者副本之前首領發生崩憤。

all意味著首領在返回確認或錯誤響應之前，會等待所有同步副本都收到悄息。這是最保險的做一一生產者也會一直重試直到訊息被成功提交。不過這也是最慢的做法，生產者在繼續傳送其他訊息之前需要等待所有副本都收到當前的訊息。可以通過使用非同步模式和更大的批次來加快速度，但這樣做通常會降低吞吐量。

Kafka 生產過程分析

Kafka生產過程

5 Kafka生產過程分析

火力電廠生產過程

Kafka 生產過程分析

Kafka生產過程

5 Kafka生產過程分析

火力電廠生產過程

相關推薦