一、生產者發送訊息的程序

1.包裝 ProducerRecord 物件

Kafka 會將發送訊息包裝為 ProducerRecord 物件， ProducerRecord 物件包含了目標主題和要發送的內容，同時還可以指定鍵和磁區，在發送 ProducerRecord 物件前，生產者會先把鍵和值物件序列化成位元組陣列，這樣它們才能夠在網路上傳輸，

2.指定磁區

接下來，資料被傳給磁區器，如果之前已經在 ProducerRecord 物件里指定了磁區，那么磁區器就不會再做任何事情，如果沒有指定磁區，那么磁區器會根據 ProducerRecord 物件的鍵來選擇一個磁區，緊接著，這條記錄被添加到一個記錄批次里，這個批次里的所有訊息會被發送到相同的主題和磁區上，

3.放入快取

分好區的訊息不是直接被發送到服務端，而是放入了生產者的一個快取里面，在這個快取里面，多條訊息會被封裝成為一個批次（batch），默認一個批次的大小是 16K，

4.發送訊息

Sender 執行緒啟動以后會從快取里面去獲取可以發送的批次，把這些記錄批次發送到相應的 broker 上，

5.接識訓傳

服務器在收到這些訊息時會回傳一個回應，如果訊息成功寫入 Kafka，就回傳一個 RecordMetaData 物件，它包含了主題和磁區資訊，以及記錄在磁區里的偏移量，如果寫入失敗，則會回傳一個錯誤，生產者在收到錯誤之后會嘗試重新發送訊息，如果達到指定的重試次數后還沒有成功，則直接拋出例外，不再重試

二、生產者整體架構

整個生產者客戶端由兩個執行緒協調運行，這兩個執行緒分別為主執行緒和 Sender 執行緒（發送線程），在主執行緒中由 kafkaProd ucer 創建訊息，然后通過可能的攔截器、序列化器和磁區器的作用之后快取到訊息累加器（ RecordAccumulator ，也稱為訊息收集器〉中， Sender 執行緒負責從 RecordAccumulator 獲取訊息并將其發送到 Ka fka

三、序列化

生產者需要用序列化器(Serializer)把物件轉換成位元組陣列才能通過網路發送給Kaflca，而在對側，消費者需要用反序列化器(Deserializer)把從Kaflca 中收到的位元組陣列轉換成相應的物件，訊息的key和value都使用了字串，對應程式中的序列化器也使用了客戶端自帶的org.apache.kaflca. common. serialization. StringSerializer, 除了用于 String 型別的序列化器，還有ByteArray、ByteBuffer、 Bytes、 Double、Integer、 Long這幾種類型，它們都實作了org.apache.kaflca. common. serialization. Serializer介面，此介面有3個方法：

public void configure(Map<String, ?> configs, boolean isKey)
public byte[] serialize(String topic, T data)
public void close()

configure()方法用來配置當前類，serialize()方法用來執行序列化操作，而close()方法用來關閉當前的序列化器，一般情況下close()是一個空方法，如果實作了此方法，則必須確保此方法的幕等性，因為這個方法很可能會被KafkaProducer 呼叫多次，生產者使用的序列化器和消費者使用的反序列化器是需要一一對應的，如果生產者使用了某種序列化器，比如StringSerializer, 而消費者使用了另一種序列化器，比如IntegerSerializer,那么是無法決議出想要的資料的

kakfa支持配置自定義序列化：只需將KafkaProducer的value.serializer 引數設定為CompanySerializer類的全限定名即可，

四、磁區器

訊息在通過send()方法發往broker 的程序中，有可能需要經過攔截器(Interceptor)、序列化器(Serializer)和磁區器(Partitioner)的一系列作用之后才能被真正地發往 broker，攔截器一般不是必需的，而序列化器是必需的，訊息經過序列化之后就需要確定它發往的磁區，如果訊息ProducerRecord中指定了 partition欄位，那么就不需要磁區器的作用，因為par巨巨on代表的就是所要發往的磁區號，如果訊息ProducerRecord中沒有指定par巨巨on欄位，那么就需要依賴磁區器，根據key 這個欄位來計算 partition 的值，磁區器的作用就是為訊息分配磁區，

Kafka 中提供的默認磁區器是org.apache.kafka.clients.producer.intemals.DefaultPartitioner, 它實作了org.apache.kafka.clients.producer.Partitioner 介面，這個介面中定義了2個方法，具體如下所示，

public int partition(S七ring topic, Object key, byte[] keyBytes,
Object value, byte[] valueBytes, Cluster cluster); 
public void close()

其中 partition （）方法用來計算磁區號，回傳值為 int 型別， partition （）方法中的引數分別表示主題、鍵、序列化后的鍵、值、序列后的值，以及集群的元資料資訊，通過這些資訊可以實現功能豐富的磁區器，c lose() 方法在關閉磁區器的時候用來回收一些資源

在默認磁區器 DefaultPartitioner 的實作中， close() 是空方法，而在 partition （）方法中定義了主要的磁區分配邏輯如果 ke 不為 null ，那么默認的磁區器會對 key 進行哈希（采用 MurmurHash2 演算法，具備高運算性能及低碰撞率），最終根據得到哈希值來算磁區號，擁有相同 key 的訊息會被寫入同一個磁區如果 key為 null ，那么訊息將會以輪詢的方式發往主題內的各個可用磁區，

如果key不為null,那么磁區號會是所有磁區中的任意一個，如果為null則僅會為可用磁區中的任意一個

除了使用 Kafka 提供的默認磁區器進行磁區分配，還可以使用自定義的磁區器，只需同 DefaultPartitioner 一樣實 Partitioner 介面即可，默認磁區器在 key null 時不會選擇非可用的磁區，我們可以通過自定義的分區器 DemoPartitioner 打破這限制

五、生產者攔截器

攔截器（ Interceptor ）是早在 Kafka 0.10.0.0 中就已經引入的個功能， Kafka 一共有兩種攔截器生產者攔截器和消費者攔截器

生產者攔截器既可以用來在訊息發送前一些準備作業比如按照某個規則過率不符合要的消息、修改訊息的內容等，也可以用來在發送回調邏輯前做一些定制化的需求，比如統計類作業，

生產者攔截器使用很方便，主要是自定義實作 org apache.kafka. clients . producer. Producerlnterceptor 介面， Producer Interceptor 口中包含三個方法

public ProducerRecord<K, V> onSend (ProducerRecord<K, V> record ); 
public void onAcknowledgement(RecordMetadata metadata , Exception exception ); 
public void close() ;

KafkaProducer 在將訊息序列化和計算磁區前會調生產者攔截器 onSend（）方法來對消息進行相應定制化操作，一般來說最好不要修改訊息 ProducerRecord 的topic和partition 等資訊，如果要修改，則需確保對其有準確的判斷，否則會與預想的效果出現偏差，比如修改 key 不僅會影響磁區的計算，同樣會影響 broker 端日志壓縮（ Log Compaction) 的功能

KafkaProducer 會在訊息被應答（ A cknowledgement ）之前或訊息發送失敗時呼叫生產者攔截器的 onAcknowledgement （）方法，優先于用戶設定的 Ca llback 之前執行，這個方法運行在 Producer I/O 執行緒中，所以這個方法中實作的代碼邏輯越簡單越好則會影響訊息的發送速度，

close （）方法主要用于在關閉攔截器時執行一些資源的清理作業，在這個方法中拋出的異常都會被捕獲并記錄到日志中，但并不會再向上傳遞，

六、RecordAccumulator訊息累加器（緩沖區）

主要用來快取訊息 Sender 執行緒可以批量發送，進減少網路傳輸的資源消耗以提升性能 RecordAccumulator 快取的大小可以通過生產者客戶端引數 buffer. memory 配置，默認值為 33554432B ，即 32MB 如果生產者發送訊息的速度超過發送到服務器的速度，則會導致生產者空間不足，這個時候 KafkaPro ducer send （）方法呼叫要么被阻塞，要么拋出例外，這個取決于引數 max block ms 的配置，此引數的默認值為 6 0000,及 60秒

內部結構

主執行緒中發送過來的訊息都會被迫加到 RecordAccumulator 的某個雙端佇列（ Deque ）中，在RecordAccumulator 的內部為每個磁區都維護了個雙端佇列，佇列中的內容就是 Prod uc e r Batch ，即 Deque ProducerBatch ＞，訊息寫入快取時，追加到雙端佇列的尾部： Sender 讀取訊息時，從雙端佇列的頭部讀取，注意 Producer Batch 不是 Producer Record, ProducerBatch 中可以包含一至多個 Producer Record 通俗地說， ProducerRecord 是生產者中創建的訊息，而 Producer Batch 是指一個訊息批次 ProducerRecord 會被包含在 Pro ducer Batch 中，這樣可以使字節的使用更加緊湊，與此同時，將較小的 Producer Record 湊成一個較大 ProducerBatch ，也可以減少網路請求的次數以提升整體的吞吐量 Producer Batch 和訊息的具體格式有關

七、kafka發送訊息

，發送訊息主要有三種模式：發后即忘(fire-and-forget)、同步(sync)及異步Casync)

send（）本身是異步的，但是呼叫send（）后可以通過代碼實作同步還是異步，異步：一旦訊息被保存在等待發送的訊息快取中，此方法就立即回傳，這樣并行發送多條訊息而不阻塞去等待每一條訊息的回應，當然也可以使用同步發送但是性能差，不推薦

簡單同步發送實作方法：

在呼叫send方法后直接呼叫get方法強行堵塞

 RecordMetadata metadata = producer.send(record).get();

異步實作

通常我們并不關心發送成功的情況，更多關注的是失敗的情況，因此 Kafka 提供了異步發送和回呼函式，代碼如下：

producer.send(record, new Callback() {
        @Override
        public void onCompletion(RecordMetadata metadata, Exception exception) {
            if (exception != null) {
                System.out.println("進行例外處理");
            } else {
                System.out.printf("topic=%s, partition=%d, offset=%s \n",
                        metadata.topic(), metadata.partition(), metadata.offset());
            }
        }
    });

八、重要的生產者引數

在kafka生產者中大部分的引數都有合理的默認值，一般不需要修改它們

1.acks

這個引數用來指定磁區中必須要有多少個副本收到這條訊息，之后生產者才會認為這條訊息是成功寫入的， acks 是生產者客戶端中一個非常重要的引數，它涉及訊息的可靠性和吞吐量之間的權衡acks 引數有3 種型別的值（都是字串型別）

（1）acks=1

默認值即為1 ，生產者發送訊息之后，只要磁區的 leader 副本成功寫入消息，那么它就會收到來自服務端的成功回應，如果訊息無法寫入 leader 副本，比如在 leader副本崩潰、重新選舉新的 leader 副本的程序中，那么生產者就會收到一個錯誤的回應，為了避免訊息丟失，生產者可以選擇重發訊息，如果訊息寫入 leader 副本并回傳成功回應給生產者，且在被其他 fo llo wer 副本拉取之前 leader 副本崩潰，那么此時訊息還是會丟失，因為新選舉的 leader 副本中并沒有這條對應的訊息 acks 設定為1，是訊息可靠性和吞吐量之間的折中方案

（2）acks = 0

生產者發送消息之后不需要等待任何服務端的回應，如果在訊息從發送到寫入 Kafka 的程序中出現某些例外，導致 Kafka 并沒有收到這條訊息，那么生產者也無從得知，訊息也就丟失了，在其他配置環境相同的情況下， acks 設定為0 可以達到最大的吞吐量，

（3）acks = -1或acks =all

生產者在消息發送之后，需要等待 ISR 中的所有副本都成功寫入訊息之后才能夠收到來自服務端的成功回應，在其他配置環境相同的情況下， acks 設定為 (all ）可以達到最強的可靠性，但這并不意味著訊息就一定可靠，因為 JSR 中可能只有 leader 副本，這樣就退化成了 acks=1 的情況，要獲得更高的訊息可靠性需要配合 min.insync.replicas 引數的聯動

注意 acks 引數配置的值是一個字串型別，而不是整數型別

2.max.request.size

該引數用于控制生產者發送的請求大小，它可以指發送的單個訊息的最大值，kafka默認的發送一條訊息的大小是1M

3.retries 和 retry.backo.ms

發生錯誤后，訊息重發的次數，如果達到設定值，生產者就會放棄重試并回傳錯誤，默認是0，即在發生例外的時候不進行任何重試動作，訊息在從生產者發出到成功寫入服務器之前可能發生一些臨時性的例外，比如網路抖動、 le der 副本的選舉等，這種例外往往是可以自行恢復的，生產者可以通過配置 retries 大于0值，以此通過內部重試來恢復而不是一昧地將例外拋給生產者的應用程式，但是不是所有例外都能處理，比如超過訊息最大值的例外

retry.backoff.ms用來設定兩次重試之間的間隔

4.compression.type

這個引數用來指定訊息的壓縮方式，默認值為“ none ”，即默認情況下，訊息不會被壓縮，該引數還可以配置為“ gzip snappy 和“ z4 對訊息進行壓縮可以極大地減少網路傳輸、降低網路I/O，從而提高整體的性能，訊息壓縮是種使用時間換間的優化方式，如果對時延有一定的要求，則不推薦對訊息進行壓縮

5.linge .ms

這個引數用來指定生產者發送 ProducerBatch 之前等待更多訊息（ ProducerRecord ）加入 Producer Batch 時間，默認值為，生產者客戶端會在 ProducerBatch 填滿或等待時間超過 linger.ms 值時發迭出去，增大這個引數的值會增加訊息的延遲，但是同時能提升一定的吞吐量，

6. receive.buffer.bytes & send.buffer.byte

這個引數用來設定 Socket 接收訊息緩沖區（ SO RE CBUF ）的大小(滑動視窗協議)，默認值接收視窗為 32KB，發生視窗為128KB，如果設定為 -1 ，則使用作業系統的默認值，如果 Producer Kafka 于不同的機房則可以適地調大這個引數值

7.timeout.ms, request.timeout.ms & metadata.fetch.timeout.ms

timeout.ms 指定了 borker 等待同步副本回傳訊息的確認時間；
request.timeout.ms 指定了生產者在發送資料時等待服務器回傳回應的時間；
metadata.fetch.timeout.ms 指定了生產者在獲取元資料（比如磁區首領是誰）時等待服務器回傳回應的時間，

轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/275130.html

標籤：其他

上一篇：K8S基礎及單節點服務的部署(etcd)

下一篇：Kubernetes（k8s）基礎簡介

kafka--生產者詳解