怎么解析Kafka Producer

本篇文章為大家展示了怎么解析Kafka Producer，內容簡明扼要并且容易理解，絕對能使你眼前一亮，通過這篇文章的詳細介紹希望你能有所收獲。

Kafka Producer在發送消息大致有以下流程：

1. 首先將消息封裝在ProducerRecord中，并且序列化

2. 將序列化后的消息發送給partitioner，partitioner主要用來確定消息發往哪個分區，默認的分區策略是輪詢，如果消息有key，具有相同key的消息可以被發往同一分區，Kafka Producer也允許用戶直接指定要發往的分區

3. Producer有一個專門的Sender線程會從緩沖區獲取消息，計算將要發往的Leader Broker然后進行發送。

必需參數

bootstrap.servers

指定一組host:port鍵值對，用于連接kafka broker節點，producer可以通過該參數發現Kafka集群中的所有broker，因此可以指定部分節點。

key.serializer

發往kafka的的消息都需要被序列化成二進制字節數組，該參數指定的是消息key的序列化格式，如果指定的StringSerializer，表明是將一個String字符串序列化成二進制字節數組。

目前支持的序列化格式有：

• ByteArraySerializer：啥都不用做，本質就是個二進制字節數組

• ByteBufferSerializer：序列化ByteBuffer

• ByteSerializer：序列化Kafka自定義的Bytes類型

• DoubleSerializer：序列化Double類型

• IntegerSerializer：序列化Integer類型

• LongSerializer：序列化Long類型

• StringSerializer：序列化String類型

Kafka支持自定義序列化類型，只需要實現org.apache.kafka.common.serialization.Serializer接口。

value.serializer

指定消息內容的序列化格式

主要參數

acks

該參數主要用來控制消息的持久性，該參數主要有三個值：0、1、all(-1)

• 0：表示Producer完全不用考慮broker的響應，發送完一條消息后立即開啟下一條消息的發送，不用理會leader broker的響應，該設置吞吐量很高，但消息容易丟失

• all(-1)：表示發送消息時，leader broker不僅會將消息寫入本地日志，同時還會等待ISR中所有副本(ISR副本數量由broker端的min.insync.replicas控制)都成功寫入他們的本地日志后，才發送響應結果給producer，該設置吞吐量最低，但消息持久性最高

• 1：producer發送消息后，leader broker進將消息成功寫入本地日志后便返回響應給producer

buffer.memory

該參數用于指定producer發送消息緩沖區的大小，單位字節，默認值32MB。增加該參數的值可以提升吞吐量，但會增加延時。

compression.type

該參數用來設置是否開啟消息壓縮，默認值為none，目前Kafka支持GZIP、Snappy和LZ4。開啟壓縮后，可以降低網絡I/O從而提升吞吐量，同時會增加CPU的壓力。

retries

Producer在發送消息的時候有可能因為網絡抖動從而發送失敗，這種失敗都是可以重試解決，retries參數決定了Producer內部的重試次數。

目前可以重試的異常主要有以下幾種：

• LeaderNotAvailableException：分區副本Leader不可用，通常出現在Leader選舉期間

• NotControllerException：controller不可用，通常出現在controller選舉期間

• NetworkException：網絡瞬時故障導致的異常

重試有可能會導致以下問題：

• 消息重復發送：kafka 0.11.0.0以前的版本需要用戶對consumer進行去重處理，但在此版本以后可以在broker端將參數enable.idempotence設置為true，便可以實現消息的去重，此參數可以保證消息的冪等性

• 消息亂序：Producer提供了max.in.flight.request.per.connect可以將該參數設置為1，表示Producer同一時刻只能發送一個請求

batch.size

Producer會將發往相同分區的消息進行匯總，當到達的該參數的設置值時，便將消息一同發往Leader Broker。

該參數的默認值是16KB，增加該參數可以提高吞吐量，但是相應的會增加消息的延遲

linger.ms

當消息無法滿足batch size，但是超過了linger.ms的值時，也會將消息一同發往Leader Broker。

該參數默認值是0，表示不關心是否填滿batch.size立即發送。

max.request.size

用于控制Producer發送請求的大小，默認值是1048576字節

request.timeout.ms

Producer發送請求給broker以后，broker需要在規定時間內返回響應，否則Producer會拋出TimemoutException需要客戶端去處理。

該參數的默認值是30s。

多線程處理

KafkaProducer是線程安全的。

上述內容就是怎么解析Kafka Producer，你們學到知識或技能了嗎？如果還想學到更多技能或者豐富自己的知識儲備，歡迎關注億速云行業資訊頻道。