[[409180]]

kafka 是單條發(fā)送還是批量發(fā)送消息?

kafka 怎么做到單條發(fā)送?

kafka 發(fā)送消息是順序的嗎?

生產(chǎn)者什么情況下可能會(huì)頻繁FullGC?

消息發(fā)送的邏輯

上帝視角來(lái)看消息發(fā)送的流程。

生產(chǎn)者的設(shè)計(jì)

消費(fèi)發(fā)送機(jī)制：

1)序列化器：序列化消息對(duì)象轉(zhuǎn)成字節(jié)數(shù)組，然后通過(guò)網(wǎng)絡(luò)傳輸。

2)分區(qū)器：計(jì)算消息發(fā)往的具體分區(qū);如果顯示指定了partition，便不會(huì)走分區(qū)器。

3)消息緩沖池：客戶端的消息緩沖池，默認(rèn)大小32M，見(jiàn)參數(shù)buffer.memory。

4)批量發(fā)送：緩沖池中消息會(huì)按batch分批次發(fā)送，默認(rèn)批次大小16KB，見(jiàn)參數(shù)batch.size。

負(fù)載均衡設(shè)計(jì)：

由于消息topic 由多個(gè)partition 組成，且partition 會(huì)均衡分布到不同broker 上。因此，為了有效利用broker 集群的性能，提高消息的吞吐量，producer 可以通過(guò)隨機(jī)或者h(yuǎn)ash 等方式，將消息平均發(fā)送到多個(gè)partition 上，以實(shí)現(xiàn)負(fù)載均衡。

分區(qū)策略：

1. 輪詢策略，默認(rèn)策略
2. 隨機(jī)策略，實(shí)際表現(xiàn)來(lái)看，它要遜于輪詢策略
3. 按消息鍵保序策略，一旦消息被定義了 Key，那么你就可以保證同一個(gè) Key 的所有消息都進(jìn)入到相同的分區(qū)里面，由于每個(gè)分區(qū)下的消息處理都是有順序的。

KafkaProducer

源碼

//客戶端ID。在創(chuàng)建 KafkaProducer 時(shí)可通過(guò) client.id 定義 clientId，如果未指定，則默認(rèn) producer- seq，seq 在進(jìn)程內(nèi)遞增，強(qiáng)烈建議客戶端顯示指定 clientId。 
 private final String clientId; 
   //度量的相關(guān)存儲(chǔ)容器，例如消息體大小、發(fā)送耗時(shí)等與監(jiān)控相關(guān)的指標(biāo)。 
    final Metrics metrics; 
    //分區(qū)負(fù)載均衡算法，通過(guò)參數(shù) partitioner.class 指定。 
    private final Partitioner partitioner; 
    //調(diào)用 send 方法發(fā)送的最大請(qǐng)求大小，包括 key、消息體序列化后的消息總大小不能超過(guò)該值。通過(guò)參數(shù) max.request.size 來(lái)設(shè)置。 
    private final int maxRequestSize; 
    //生產(chǎn)者緩存所占內(nèi)存的總大小，通過(guò)參數(shù) buffer.memory 設(shè)置。 
    private final long totalMemorySize; 
    //元數(shù)據(jù)信息，例如 topic 的路由信息，由 KafkaProducer 自動(dòng)更新。 
    private final Metadata metadata; 
    //消息記錄累積器 
    private final RecordAccumulator accumulator; 
    //用于封裝消息發(fā)送的邏輯，即向 broker 發(fā)送消息的處理邏輯。 
    private final Sender sender; 
    //用于消息發(fā)送的后臺(tái)線程，一個(gè)獨(dú)立的線程，內(nèi)部使用 Sender 來(lái)向 broker 發(fā)送消息。 
    private final Thread ioThread; 
    //壓縮類(lèi)型，默認(rèn)不啟用壓縮，可通過(guò)參數(shù) compression.type 配置。可選值：none、gzip、snappy、lz4、zstd。 
    private final CompressionType compressionType; 
    //錯(cuò)誤信息收集器，當(dāng)成一個(gè) metrics，用來(lái)做監(jiān)控的。 
    private final Sensor errors; 
    //用于獲取系統(tǒng)時(shí)間或線程睡眠等。 
    private final Time time; 
    //用于對(duì)消息的 key 進(jìn)行序列化。 
    private final ExtendedSerializer<K> keySerializer; 
    //Serializer< V> valueSerializer 
    private final ExtendedSerializer<V> valueSerializer; 
    //生產(chǎn)者的配置信息。 
    private final ProducerConfig producerConfig; 
    //最大阻塞時(shí)間，當(dāng)生產(chǎn)者使用的緩存已經(jīng)達(dá)到規(guī)定值后，此時(shí)消息發(fā)送會(huì)阻塞，通過(guò)參數(shù) max.block.ms 來(lái)設(shè)置最多等待多久。 
    private final long maxBlockTimeMs; 
   //配置控制客戶機(jī)等待請(qǐng)求響應(yīng)的最長(zhǎng)時(shí)間。如果在超時(shí)超時(shí)之前沒(méi)有收到響應(yīng)，客戶端將在需要時(shí)重新發(fā)送請(qǐng)求，或者在重試耗盡時(shí)失敗請(qǐng)求。 
    private final int requestTimeoutMs; 
    //生產(chǎn)者端的攔截器，在消息發(fā)送之前進(jìn)行一些定制化處理。 
    private final ProducerInterceptors<K, V> interceptors; 
    //維護(hù) api 版本的相關(guān)元信息，該類(lèi)只能在 kafka 內(nèi)部使用。 
    private final ApiVersions apiVersions; 
    //kafka 消息事務(wù)管理器。 
    private final TransactionManager transactionManager; 
    //kafka 生產(chǎn)者事務(wù)上下文環(huán)境初始結(jié)果。 
    private TransactionalRequestResult initTransactionsResult; 

KafkaProducer 具有如下特征：

1. KafkaProducer 是線程安全的，可以被多個(gè)線程交叉使用。
2. KafkaProducer 內(nèi)部包含一個(gè)緩存池，存放待發(fā)送消息，即 ProducerRecord 隊(duì)列，與此同時(shí)會(huì)開(kāi)啟一個(gè)IO線程將 ProducerRecord 對(duì)象發(fā)送到 Kafka 集群。
3. KafkaProducer 的消息發(fā)送 API send 方法是異步，只負(fù)責(zé)將待發(fā)送消息 ProducerRecord 發(fā)送到緩存區(qū)中，立即返回，并返回一個(gè)結(jié)果憑證 Future。

acks 參數(shù)的作用

KafkaProducer 提供了一個(gè)核心參數(shù) acks 用來(lái)定義消息“已提交”的條件(標(biāo)準(zhǔn))，就是 Broker 端向客戶端承偌已提交的條件，可選值如下：

1. 0：只要調(diào)用 KafkaProducer 的 send 方法返回后即認(rèn)為成功
2. all 或 -1：表示消息不僅需要 Leader 節(jié)點(diǎn)已存儲(chǔ)該消息，并且要求其副本(準(zhǔn)確的來(lái)說(shuō)是 ISR 中的節(jié)點(diǎn))全部存儲(chǔ)才認(rèn)為已提交，才向客戶端返回提交成功。這是最嚴(yán)格的持久化保障，當(dāng)然性能也最低。
3. 1：表示消息只需要寫(xiě)入 Leader 節(jié)點(diǎn)后就可以向客戶端返回提交成功。

retries 參數(shù)的作用

kafka 在生產(chǎn)端提供的另外一個(gè)核心屬性，用來(lái)控制消息在發(fā)送失敗后的重試次數(shù)，設(shè)置為 0 表示不重試，重試就有可能造成消息在發(fā)送端的重復(fù)。從消息發(fā)送接口來(lái)看：

Future<RecordMetadata> send(ProducerRecord<K, V> record);Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback); 

從上面的 API 可以得知，用戶在使用 KafkaProducer 發(fā)送消息時(shí)，首先需要將待發(fā)送的消息封裝成 ProducerRecord，返回的是一個(gè) Future 對(duì)象，典型的 Future 設(shè)計(jì)模式。

Kafka 消息追加流程

KafkaProducer 的 send 方法，并不會(huì)直接向 broker 發(fā)送消息，kafka 將消息發(fā)送異步化，即分解成兩個(gè)步驟，send 方法的職責(zé)是將消息追加到內(nèi)存中(分區(qū)的緩存隊(duì)列中)，然后會(huì)由專(zhuān)門(mén)的 Send 線程異步將緩存中的消息批量發(fā)送到 Kafka Broker 中。

主要的方法在KafkaProducer#doSend

將消息追加到生產(chǎn)者的發(fā)送緩存區(qū)，其實(shí)現(xiàn)類(lèi)為：RecordAccumulator。我們先來(lái)看一下 Kafka 一條消息寫(xiě)到內(nèi)存的流程圖：

Sender線程

到此為止，我們看到，當(dāng)我們調(diào)用send 方法的時(shí)候，其實(shí)只是發(fā)送到了 生產(chǎn)者客戶端的服務(wù)內(nèi)存中。還沒(méi)有到Broker。Kafka producer 客戶端后臺(tái)會(huì)啟動(dòng)一個(gè)線程不停的輪詢消息批次存放的區(qū)域，把消息發(fā)送給Broker。

消息批次的內(nèi)存結(jié)構(gòu)和分配

根據(jù)上面的源碼我們可以了解到，每一個(gè)ProducerBatch 是一塊 大小為batch.size 字節(jié)大小的內(nèi)存。而且用到了池化技術(shù)。

緩沖池的內(nèi)存持有類(lèi)是 BufferPool，我們先來(lái)看下 BufferPool 都有哪些成員：

public class BufferPool { 
  // 總的內(nèi)存大小 
  private final long totalMemory; 
  // 每個(gè)內(nèi)存塊大小，即 batch.size 
  private final int poolableSize; 
  // 申請(qǐng)、歸還內(nèi)存的方法的同步鎖 
  private final ReentrantLock lock; 
  // 空閑的內(nèi)存塊 
  private final Deque<ByteBuffer> free; 
  // 需要等待空閑內(nèi)存塊的事件 
  private final Deque<Condition> waiters; 
  /** Total available memory is the sum of nonPooledAvailableMemory and the number of byte buffers in free * poolableSize.  */ 
  // 緩沖池還未分配的空閑內(nèi)存，新申請(qǐng)的內(nèi)存塊就是從這里獲取內(nèi)存值 
  private long nonPooledAvailableMemory; 
  // ... 
} 

從 BufferPool 的成員可看出，緩沖池實(shí)際上由一個(gè)個(gè) ByteBuffer 組成的，BufferPool 持有這些內(nèi)存塊，并保存在成員 free 中，free 的總大小由 totalMemory 作限制，而 nonPooledAvailableMemory 則表示還剩下緩沖池還剩下多少內(nèi)存還未被分配。

當(dāng) Batch 的消息發(fā)送完畢后，就會(huì)將它持有的內(nèi)存塊歸還到 free 中，以便后面的 Batch 申請(qǐng)內(nèi)存塊時(shí)不再創(chuàng)建新的 ByteBuffer，從 free 中取就可以了，從而避免了內(nèi)存塊被 JVM 回收的問(wèn)題。

創(chuàng)建內(nèi)存塊的流程如下：

歸還內(nèi)存塊的邏輯流程

如果歸還的內(nèi)存塊大小等于 batchSize，則將其清空后添加到緩沖池的 free 中，即將其歸還給緩沖池，避免了 JVM GC 回收該內(nèi)存塊。如果不等于就直接將內(nèi)存大小累加到未分配并且空閑的內(nèi)存大小值中即可，內(nèi)存就無(wú)需歸還了，等待 JVM GC 回收掉，最后喚醒正在等待空閑內(nèi)存的線程。

Java生產(chǎn)者是如何管理TCP連接的

為何采用 TCP?

Apache Kafka 的所有通信都是基于 TCP 的，而不是基于 HTTP 或其他協(xié)議。無(wú)論是生產(chǎn)者、消費(fèi)者，還是 Broker 之間的通信都是如此。

從社區(qū)的角度來(lái)看，在開(kāi)發(fā)客戶端時(shí)，人們能夠利用 TCP 本身提供的一些高級(jí)功能，比如多路復(fù)用請(qǐng)求以及同時(shí)輪詢多個(gè)連接的能力。

TCP 的多路復(fù)用請(qǐng)求會(huì)在一條物理連接上創(chuàng)建若干個(gè)虛擬連接，每個(gè)虛擬連接負(fù)責(zé)流轉(zhuǎn)各自對(duì)應(yīng)的數(shù)據(jù)流。其實(shí)嚴(yán)格來(lái)說(shuō)，TCP 并不能多路復(fù)用，它只是提供可靠的消息交付語(yǔ)義保證，比如自動(dòng)重傳丟失的報(bào)文。

而且目前已知的 HTTP 庫(kù)在很多編程語(yǔ)言中都略顯簡(jiǎn)陋。

何時(shí)創(chuàng)建 TCP 連接?

TCP 連接是在創(chuàng)建 KafkaProducer 實(shí)例時(shí)建立的 ，在創(chuàng)建 KafkaProducer 實(shí)例時(shí)，生產(chǎn)者應(yīng)用會(huì)在后臺(tái)創(chuàng)建并啟動(dòng)一個(gè)名為 Sender 的線程，該 Sender 線程開(kāi)始運(yùn)行時(shí)首先會(huì)創(chuàng)建與 Broker 的連接。

Properties properties = new Properties(); 
properties.put("bootstrap.servers", "localhost:9092"); 
properties.put("key.serializer", StringSerializer.class.getName()); 
properties.put("value.serializer", StringSerializer.class.getName()); 
// try-with-resources 
// 創(chuàng)建KafkaProducer實(shí)例時(shí)，會(huì)在后臺(tái)創(chuàng)建并啟動(dòng)Sender線程，Sender線程開(kāi)始運(yùn)行時(shí)首先會(huì)創(chuàng)建與Broker的TCP連接 
try (Producer<String, String> producer = new KafkaProducer<>(properties)) { 
    ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, KEY, VALUE); 
    Callback callback = (metadata, exception) -> { 
    }; 
    producer.send(record, callback); 
} 

1. bootstrap.servers是Producer的核心參數(shù)之一，指定了Producer啟動(dòng)時(shí)要連接的Broker地址
2. 如果bootstrap.servers指定了1000個(gè)Broker，那么Producer啟動(dòng)時(shí)會(huì)首先創(chuàng)建與這1000個(gè)Broker的TCP連接
3. 因此不建議把集群中所有的Broker信息都配置到bootstrap.servers中，通常配置3~4臺(tái)足夠
4. Producer一旦連接到集群中的任意一臺(tái)Broker，就能拿到整個(gè)集群的Broker信息(metadata request)

TCP 連接還可能在兩個(gè)地方被創(chuàng)建：一個(gè)是在更新元數(shù)據(jù)后，另一個(gè)是在消息發(fā)送時(shí)。

• 當(dāng)Producer更新了集群的元數(shù)據(jù)后，如果發(fā)現(xiàn)與某些Broker當(dāng)前沒(méi)有連接，那么Producer會(huì)創(chuàng)建一個(gè)TCP連接

【場(chǎng)景1】

當(dāng)Producer嘗試向不存在的主題發(fā)送消息時(shí)，Broker會(huì)告訴Producer這個(gè)主題不存在，此時(shí)Producer會(huì)發(fā)送metadata request到Kafka集群，去嘗試獲取最新的元數(shù)據(jù)信息，與集群中所有的Broker建立TCP連接。

【場(chǎng)景2】

Producer通過(guò)metadata.max.age.ms參數(shù)定期地去更新元數(shù)據(jù)信息，默認(rèn)值300000，即5分鐘。

• 當(dāng)Producer要發(fā)送消息時(shí)，Producer發(fā)現(xiàn)與目標(biāo)Broker(依賴負(fù)載均衡算法)還沒(méi)有連接，也會(huì)創(chuàng)建一個(gè)TCP連接。

何時(shí)關(guān)閉 TCP 連接?

Producer端關(guān)閉TCP連接有兩種方式：用戶主動(dòng)關(guān)閉、Kafka自動(dòng)關(guān)閉。

【用戶主動(dòng)關(guān)閉】

廣義的主動(dòng)關(guān)閉，包括用戶調(diào)用kill -9來(lái)殺掉Producer，最推薦的方式：producer.close()

【Kafka自動(dòng)關(guān)閉】

Producer端參數(shù)connections.max.idle.ms，默認(rèn)值540000，即9分鐘

如果9分鐘內(nèi)沒(méi)有任何請(qǐng)求經(jīng)過(guò)某個(gè)TCP連接，Kafka會(huì)主動(dòng)把TCP連接關(guān)閉

connections.max.idle.ms=-1會(huì)禁用這種機(jī)制，TCP連接將成為永久長(zhǎng)連接

Kafka創(chuàng)建的Socket連接都開(kāi)啟了keepalive。

【注意】

關(guān)閉TCP連接的發(fā)起方是Kafka客戶端，屬于被動(dòng)關(guān)閉的場(chǎng)景

被動(dòng)關(guān)閉的后果就是會(huì)產(chǎn)生大量的CLOSE_WAIT連接

Producer端或Client端沒(méi)有機(jī)會(huì)顯式地觀測(cè)到此TCP連接已被中斷

總結(jié)

現(xiàn)在我們可以回答開(kāi)頭的3個(gè)問(wèn)題了。

1、kafka 是單條發(fā)送還是批量發(fā)送消息?

正常情況下都是批量發(fā)送的。封裝成一個(gè)ProducerBatch 發(fā)送。

2.kafka 怎么做到單條發(fā)送?

只能設(shè)置單生產(chǎn)者單線程同步調(diào)用send 方法。

3.kafka 發(fā)送消息是順序的嗎?

不是的，如果需求順序必須設(shè)置key，并且是生產(chǎn)者是單線程的。

4.生產(chǎn)者什么情況下可能會(huì)頻繁FullGC?

如果你的消息大小比 batchSize 還要大，則不會(huì)從 free 中循環(huán)獲取已分配好的內(nèi)存塊，而是重新創(chuàng)建一個(gè)新的 ByteBuffer，并且該 ByteBuffer 不會(huì)被歸還到緩沖池中(JVM GC 回收)，如果此時(shí) nonPooledAvailableMemory 比消息體還要小，還會(huì)將 free 中空閑的內(nèi)存塊銷(xiāo)毀(JVM GC 回收)，以便緩沖池中有足夠的內(nèi)存空間提供給用戶申請(qǐng)，這些動(dòng)作都會(huì)導(dǎo)致頻繁 GC 的問(wèn)題出現(xiàn)。

因此，需要根據(jù)業(yè)務(wù)消息的大小，適當(dāng)調(diào)整 batch.size 的大小，避免頻繁 GC。