https://zhuanlan.zhihu.com/p/367683572java
@TOCnode
show the code.git
public class KafkaProducerDemo { public static void main(String[] args) { // step 1: 設置必要參數 Properties config = new Properties(); config.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092,127.0.0.1:9093"); config.setProperty(ProducerConfig.ACKS_CONFIG, "-1"); config.setProperty(ProducerConfig.RETRIES_CONFIG, "3"); // step 2: 建立KafkaProducer KafkaProducer<String, String> producer = new KafkaProducer<>(config); // step 3: 構造要發送的消息 String topic = "kafka-source-code-demo-topic"; String key = "demo-key"; String value = "村口老張頭: This is a demo message."; ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value); // step 4: 發送消息 Future<RecordMetadata> future = producer.send(record); } }
代碼中涉及的幾個配置:github
KafkaProducer兩個模板參數指定了消息的key和value的類型算法
String topic = "kafka-source-code-demo-topic";
String key = "demo-key";
key用來決定目標Partition,這個下文細聊。apache
String value = "村口老張頭: This is a demo message.";
這是待發送的消息內容,傳遞業務數據。bootstrap
Future<RecordMetadata> future = producer.send(record);
KafkaProducer中各種send方法均返回Future,並不會直接返回發送結果,其緣由即是線程模型設計。api
這裏主要存在兩個線程:主線程 和 Sender線程。主線程即調用KafkaProducer.send方法的線程。當send方法被調用時,消息並無真正被髮送,而是暫存到RecordAccumulator。Sender線程在知足必定條件後,會去RecordAccumulator中取消息併發送到Kafka Server端。
那麼爲啥不直接在主線程就把消息發送出去,非得搞個暫存呢?爲了Kafka的目標之一——高吞吐。具體而言有兩點好處:緩存
先給個總體流程圖,而後我們再逐步分析。微信
這裏列出KafkaProducer的核心屬性。至於所有屬性說明,可參考個人"註釋版Kafka源碼":https://github.com/Hao1296/kafka |
字段名 | 字段類型 | 說明 |
---|---|---|---|
clientId | String | 生產者惟一標識 | |
partitioner | Partitioner | 分區選擇器 | |
metadata | Metadata | Kafka集羣元數據 | |
accumulator | RecordAccumulator | 消息緩存器 | |
sender | Sender | Sender線程業務邏輯封裝,繼承Runnable | |
ioThread | Thread | Sender線程對應的線程對象 | |
interceptors | ProducerInterceptors | 消息攔截器,下文會說明 |
ProducerInterceptors,消息攔截器集合,維護了多個ProducerInterceptor對象。用於在消息發送前對消息作額外的業務操做。使用時可按以下方式設置:
Properties config = new Properties(); // interceptor.classes config.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "com.kafka.demo.YourProducerInterceptor,com.kafka.demo.InterceptorImpl2"); KafkaProducer<String, String> producer = new KafkaProducer<>(config);
ProducerInterceptor自己是一個接口:
public interface ProducerInterceptor<K, V> extends Configurable { ProducerRecord<K, V> onSend(ProducerRecord<K, V> record); void onAcknowledgement(RecordMetadata metadata, Exception exception); void close(); }
其中,onAcknowledgement是獲得Server端正確響應時的回調,後面再細說。onSend是消息在發送前的回調,可在這裏作一些消息變動邏輯(如加減字段等)。輸入原始消息,輸出變動後的消息。KafkaProducer的send方法第一步就是執行ProducerInterceptor:
@Override public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) { // intercept the record, which can be potentially modified; // this method does not throw exceptions // 關注這裏 ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record); return doSend(interceptedRecord, callback); } // 該send方法重載核心邏輯還是上面的send方法 @Override public Future<RecordMetadata> send(ProducerRecord<K, V> record) { return send(record, null); }
接上文,ProducerInterceptors執行完畢後會直接調用doSend方法執行發送相關的邏輯。到這爲止有個問題,咱們並不知道目標Topic下有幾個Partition,分別分佈在哪些Broker上;故,咱們也不知道消息該發給誰。因此,doSend方法第一步就是搞清楚消息集羣結構,即獲取集羣元數據:
private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) { TopicPartition tp = null; try { throwIfProducerClosed(); ClusterAndWaitTime clusterAndWaitTime; try { // 獲取集羣元數據 clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs); } catch (KafkaException e) { if (metadata.isClosed()) throw new KafkaException("Producer closed while send in progress", e); throw e; } ... ... }
waiteOnMetadata方法內部大致分爲2步:
private ClusterAndWaitTime waitOnMetadata(String topic, Integer partition, long maxWaitMs) throws InterruptedException { // 第1步, 判斷是否已經有了對應topic&partition的元數據 Cluster cluster = metadata.fetch(); Integer partitionsCount = cluster.partitionCountForTopic(topic); if (partitionsCount != null && (partition == null || partition < partitionsCount)) // 若已存在, 則直接返回 return new ClusterAndWaitTime(cluster, 0); // 第2步, 獲取元數據 do { ... ... // 2.1 將目標topic加入元數據對象 metadata.add(topic); // 2.3 將元數據needUpdate字段置爲true, 並返回當前元數據版本 int version = metadata.requestUpdate(); // 2.4 喚醒Sender線程 sender.wakeup(); // 2.5 等待已獲取的元數據版本大於version時返回, 等待時間超過remainingWaitMs時拋異常 try { metadata.awaitUpdate(version, remainingWaitMs); } catch (TimeoutException ex) { throw new TimeoutException( String.format("Topic %s not present in metadata after %d ms.", topic, maxWaitMs)); } // 2.6 檢查新版本元數據是否包含目標partition; // 若包含, 則結束循環; 若不包含, 則進入下一個迭代, 獲取更新版本的元數據 cluster = metadata.fetch(); ...... partitionsCount = cluster.partitionCountForTopic(topic); } while (partitionsCount == null || (partition != null && partition >= partitionsCount)); return new ClusterAndWaitTime(cluster, elapsed); }
咱們看到,waitOnMetadata的思想也和簡單,即:喚醒Sender線程來更新元數據,而後等待元數據更新完畢。至於Sender線程是如何更新元數據的,放到下文詳解。
這一步是用經過"key.serializer"和"value.serializer"兩個配置指定的序列化器分別來序列化key和value
private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) { ..... // key序列化 byte[] serializedKey; try { serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key()); } catch (ClassCastException cce) { throw new SerializationException("Can't convert key of class " + record.key().getClass().getName() + " to class " + producerConfig.getClass(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG).getName() + " specified in key.serializer", cce); } // value序列化 byte[] serializedValue; try { serializedValue = valueSerializer.serialize(record.topic(), record.headers(), record.value()); } catch (ClassCastException cce) { throw new SerializationException("Can't convert value of class " + record.value().getClass().getName() + " to class " + producerConfig.getClass(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG).getName() + " specified in value.serializer", cce); } ...... }
Kafka內置了幾個Serializer,若是須要的話,諸君也能夠自定義:
org.apache.kafka.common.serialization.StringSerializer;
org.apache.kafka.common.serialization.LongSerializer;
org.apache.kafka.common.serialization.IntegerSerializer;
org.apache.kafka.common.serialization.ShortSerializer;
org.apache.kafka.common.serialization.FloatSerializer;
org.apache.kafka.common.serialization.DoubleSerializer;
org.apache.kafka.common.serialization.BytesSerializer;
org.apache.kafka.common.serialization.ByteBufferSerializer;
org.apache.kafka.common.serialization.ByteArraySerializer;
到這裏,咱們已經有了Topic相關的元數據,但也很快遇到了一個問題:Topic下可能有多個Partition,做爲生產者,該將待發消息發給哪一個Partition?這就用到了上文提到過的KafkaProducer的一個屬性——partitioner。
private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) { ...... // 肯定目標Partition int partition = partition(record, serializedKey, serializedValue, cluster); ...... } private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) { // 若ProducerRecord中強制指定了partition, 則以該值爲準 Integer partition = record.partition(); // 不然調用Partitioner動態計算對應的partition return partition != null ? partition : partitioner.partition( record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster); }
在建立KafkaProducer時,能夠經過"partitioner.class"配置來指定Partitioner的實現類。若未指定,則使用Kafka內置實現類——DefaultPartitioner。DefaultPartitioner的策略也很簡單:若未指定key,則在Topic下多個Partition間Round-Robin;若指定了key,則經過key來hash到一個partition。
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) { // 若未指定key 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; } }
RecordAccumulator做爲消息暫存者,其思想是將目的地Partition相同的消息放到一塊兒,並按必定的"規格"(由"batch.size"配置指定)劃分紅多個"批次"(ProducerBatch),而後以批次爲單位進行數據壓縮&發送。示意圖以下:![]() RecordAccumulator核心屬性以下: |
字段名 | 字段類型 | 說明 |
---|---|---|---|
batches | ConcurrentMap<TopicPartition, Deque<ProducerBatch>> | 按Partition維度存儲消息數據,<br>即上文示意圖描述的結構 | |
compression | CompressionType | 數據壓縮算法 |
RecordAccumulator有兩個核心方法,分別對應"存"和"取":
/** * 主線程會調用此方法追加消息 */ public RecordAppendResult append(TopicPartition tp, long timestamp, byte[] key, byte[] value, Header[] headers, Callback callback, long maxTimeToBlock) throws InterruptedException; /** * Sender線程會調用此方法提取消息 */ public Map<Integer, List<ProducerBatch>> drain(Cluster cluster, Set<Node> nodes, int maxSize, long now);
在分析Sender線程業務邏輯前,先來講說通訊基礎類。
NetworkClient有兩個核心方法:
public void send(ClientRequest request, long now); public List<ClientResponse> poll(long timeout, long now);
其中,send方法頗有迷惑性。乍一看,以爲其業務邏輯是將request同步發送出去。然而,send方法其實並不實際執行向網絡端口寫數據的動做,只是將請求"暫存"起來。poll方法纔是實際執行讀寫動做的地方(NIO)。當請求的目標channel可寫時,poll方法會實際執行發送動做;當channel有數據可讀時,poll方法讀取響應,並作對應處理。
NetworkClient有一個核心屬性:
/* 實際實現類爲 org.apache.kafka.common.network.Selector */ private final Selectable selector;
send和poll方法都是經過selector來完成的:
public void send(ClientRequest request, long now) { doSend(request, false, now); } private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long now) { ... ... doSend(clientRequest, isInternalRequest, now, builder.build(version)); } private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long now, AbstractRequest request) { ... ... selector.send(send); } public List<ClientResponse> poll(long timeout, long now) { ... ... this.selector.poll(Utils.min(timeout, metadataTimeout, defaultRequestTimeoutMs)); ... ... }
org.apache.kafka.common.network.Selector 內部則經過 java.nio.channels.Selector 來實現。
值得關注的一點是,NetworkClient的poll方法在調用Selector的poll方法前還有段業務邏輯:
// 在selector.poll前有此行邏輯 long metadataTimeout = metadataUpdater.maybeUpdate(now); try { this.selector.poll(Utils.min(timeout, metadataTimeout, defaultRequestTimeoutMs)); } catch (IOException e) { log.error("Unexpected error during I/O", e); }
metadataUpdater.maybeUpdate能夠看出是爲元數據更新服務的。其業務邏輯是:判斷是否須要更新元數據;若須要,則經過NetworkClient.send方法將MetadataRequest也加入"暫存",等待selector.poll中被實際發送出去。
KafkaProducer中,和Sender線程相關的有兩個屬性: |
字段名 | 字段類型 | 說明 |
---|---|---|---|
ioThread | Thread | Sender線程實例 | |
sender | Sender | Runnable實例,爲Sender線程的具體業務邏輯 |
在KafkaProducer的構造函數中被建立:
KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer, Metadata metadata, KafkaClient kafkaClient) { ... ... this.sender = new Sender(logContext, client, this.metadata, this.accumulator, maxInflightRequests == 1, config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG), acks, retries, metricsRegistry.senderMetrics, Time.SYSTEM, this.requestTimeoutMs, config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG), this.transactionManager, apiVersions); String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId; this.ioThread = new KafkaThread(ioThreadName, this.sender, true); this.ioThread.start(); ... ... }
Sender線程的業務邏輯也很清晰:
public void run() { log.debug("Starting Kafka producer I/O thread."); // 主循環 while (running) { try { run(time.milliseconds()); } catch (Exception e) { log.error("Uncaught error in kafka producer I/O thread: ", e); } } log.debug("Beginning shutdown of Kafka producer I/O thread, sending remaining records."); // 下面是關閉流程 // okay we stopped accepting requests but there may still be // requests in the accumulator or waiting for acknowledgment, // wait until these are completed. while (!forceClose && (this.accumulator.hasUndrained() || this.client.inFlightRequestCount() > 0)) { try { run(time.milliseconds()); } catch (Exception e) { log.error("Uncaught error in kafka producer I/O thread: ", e); } } if (forceClose) { // We need to fail all the incomplete batches and wake up the threads waiting on // the futures. log.debug("Aborting incomplete batches due to forced shutdown"); this.accumulator.abortIncompleteBatches(); } try { this.client.close(); } catch (Exception e) { log.error("Failed to close network client", e); } log.debug("Shutdown of Kafka producer I/O thread has completed."); }
主循環中僅僅是不斷調用另外一個run重載,該重載的核心業務邏輯以下:
void run(long now) { ... ... // 1. 發送請求,並肯定下一步的阻塞超時時間 long pollTimeout = sendProducerData(now); // 2. 處理端口事件,poll的timeout爲上一步計算結果 client.poll(pollTimeout, now); }
其中,sendProducerData會調用RecordAccumulator.drain方法獲取待發送消息,而後構造ProduceRequest對象,並調用NetworkClient.send方法"暫存"。sendProducerData方法以後即是調用NetworkClient.poll來執行實際的讀寫操做。
本文分析了KafkaProducer的業務模型及核心源碼實現。才疏學淺,不必定很全面,歡迎諸君隨時討論交流。後續還會有其餘模塊的分析文章,具體可見系列文章目錄: https://zhuanlan.zhihu.com/p/367683572
微信搜索"村口老張頭",不定時推送技術文章哦~
也能夠在知乎搜索"村口老張頭"哦~