public class KafkaProducer<K, V> implements Producer<K, V> {
    public static final String NETWORK_THREAD_PREFIX = "kafka-producer-network-thread";
    // visible for testing
    @SuppressWarnings("unchecked")
    KafkaProducer(Map<String, Object> configs,Serializer<K> keySerializer,
              Serializer<V> valueSerializer,ProducerMetadata metadata,
              KafkaClient kafkaClient,ProducerInterceptors<K, V> interceptors,Time time) {
        try {
            ...
            this.sender = newSender(logContext, kafkaClient, this.metadata);
            String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
            this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
            this.ioThread.start();
            ...
            log.debug("Kafka producer started");
        } catch (Throwable t) {
           ...
        }
    }
}

/**
 * The background thread that handles the sending of produce requests to the Kafka cluster. This thread makes metadata
 * requests to renew its view of the cluster and then sends produce requests to the appropriate nodes.
 */
public class Sender implements Runnable {
    /* the state of each nodes connection */
    private final KafkaClient client; // 為 Sender 線程提供管理網(wǎng)絡(luò)連接進(jìn)行網(wǎng)絡(luò)讀寫(xiě)
    /* the record accumulator that batches records */
    private final RecordAccumulator accumulator; // 消息倉(cāng)庫(kù)累加器
    /* the metadata for the client */
    private final ProducerMetadata metadata; // 生產(chǎn)者元數(shù)據(jù)
    /* the flag indicating whether the producer should guarantee the message order on the broker or not. */
    private final boolean guaranteeMessageOrder; // 是否保證消息在 broker 端的順序性 
    /* the maximum request size to attempt to send to the server */
    private final int maxRequestSize; //發(fā)送消息最大字節(jié)數(shù)。  
    /* the number of acknowledgements to request from the server */
    private final short acks; // 生產(chǎn)者的消息發(fā)送確認(rèn)機(jī)制
    /* the number of times to retry a failed request before giving up */
    private final int retries; // 發(fā)送失敗后的重試次數(shù)，默認(rèn)為0次

    /* true while the sender thread is still running */
    private volatile boolean running; // Sender 線程是否還在運(yùn)行中     
    /* true when the caller wants to ignore all unsent/inflight messages and force close.  */
    private volatile boolean forceClose; // 是否強(qiáng)制關(guān)閉，此時(shí)會(huì)忽略正在發(fā)送中的消息。
    /* the max time to wait for the server to respond to the request*/
    private final int requestTimeoutMs; // 等待服務(wù)端響應(yīng)的最大時(shí)間,默認(rèn)30s       
    /* The max time to wait before retrying a request which has failed */
    private final long retryBackoffMs; // 失敗重試退避時(shí)間      
    /* current request API versions supported by the known brokers */
    private final ApiVersions apiVersions; // 所有 node 支持的 api 版本
    /* all the state related to transactions, in particular the producer id, producer epoch, and sequence numbers */
    private final TransactionManager transactionManager; // 事務(wù)管理，這里忽略 后續(xù)會(huì)有專門(mén)一篇講解事務(wù)相關(guān)的
    // A per-partition queue of batches ordered by creation time for tracking the in-flight batches
    private final Map<TopicPartition, List<ProducerBatch>> inFlightBatches; // 正在執(zhí)行發(fā)送相關(guān)的消息批次集合， key為分區(qū)，value是 list<ProducerBatch> 。

/**
  * The main run loop for the sender thread
  */
@Override
public void run() {
        log.debug("Starting Kafka producer I/O thread.");
        // 這里使用 volatile boolean 類型的變量 running，判斷 Sender 線程是否在運(yùn)行狀態(tài)中。
        // 1. 如果 Sender 線程在運(yùn)行狀態(tài)即 running=true，則一直循環(huán)調(diào)用 runOnce() 方法。
        while (running) {
            try {
                // 將緩沖區(qū)的消息發(fā)送到 broker。
                runOnce();
            } 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.");
        // 2. 如果（沒(méi)有強(qiáng)制關(guān)閉 && ((消息累加器中還有剩余消息待發(fā)送 || 還有等待未響應(yīng)的消息 ) || 還有事務(wù)請(qǐng)求未完成))，則繼續(xù)發(fā)送剩下的消息。
        while (!forceClose && ((this.accumulator.hasUndrained() || this.client.inFlightRequestCount() > 0) || hasPendingTransactionalRequests())) {
            try {
                // 繼續(xù)執(zhí)行將剩余的消息發(fā)送完畢
                runOnce();
            } catch (Exception e) {
                log.error("Uncaught error in kafka producer I/O thread: ", e);
            }
        }
        // 3. 對(duì)進(jìn)行中的事務(wù)進(jìn)行中斷，則繼續(xù)發(fā)送剩下的消息。
        while (!forceClose && transactionManager != null && transactionManager.hasOngoingTransaction()) {
            if (!transactionManager.isCompleting()) {
                log.info("Aborting incomplete transaction due to shutdown");
                transactionManager.beginAbort();
            }
            try {
                runOnce();
            } catch (Exception e) {
                log.error("Uncaught error in kafka producer I/O thread: ", e);
            }
        }
        // 4. 如果強(qiáng)制關(guān)閉，則關(guān)閉事務(wù)管理器、終止消息的追加并清空未完成的批次
        if (forceClose) {
            if (transactionManager != null) {
                log.debug("Aborting incomplete transactional requests due to forced shutdown");
                // 關(guān)閉事務(wù)管理器
                transactionManager.close();
            }
            log.debug("Aborting incomplete batches due to forced shutdown");
            // 終止消息的追加并清空未完成的批次
            this.accumulator.abortIncompleteBatches();
        }
        // 5. 關(guān)閉 NetworkClient
        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 a single iteration of sending
 */
void runOnce() {
        if (transactionManager != null) {
           ... //事務(wù)處理方法 后續(xù)文章專門(mén)講解 
        }
        // 1. 獲取當(dāng)前時(shí)間的時(shí)間戳。
        long currentTimeMs = time.milliseconds();
        // 2. 調(diào)用 sendProducerData 發(fā)送消息,但并非真正的發(fā)送，而是把消息緩存在 把消息緩存在 KafkaChannel 的 Send 字段里。下一篇會(huì)講解 NetworkClient。
        long pollTimeout = sendProducerData(currentTimeMs);
        // 3. 讀取消息實(shí)現(xiàn)真正的網(wǎng)絡(luò)發(fā)送
        client.poll(pollTimeout, currentTimeMs);
}

private long sendProducerData(long now) {
        // 1. 獲取元數(shù)據(jù)
        Cluster cluster = metadata.fetch();
        // get the list of partitions with data ready to send
        // 2. 獲取已經(jīng)準(zhǔn)備好的節(jié)點(diǎn)以及找不到 Leader 分區(qū)對(duì)應(yīng)的節(jié)點(diǎn)的主題
        RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);
        // if there are any partitions whose leaders are not known yet, force metadata update
        // 3. 如果主題 Leader 分區(qū)對(duì)應(yīng)的節(jié)點(diǎn)不存在，則強(qiáng)制更新元數(shù)據(jù)
        if (!result.unknownLeaderTopics.isEmpty()) {
            // 添加 topic 到?jīng)]有拉取到元數(shù)據(jù)的 topic 集合中，并標(biāo)識(shí)需要更新元數(shù)據(jù)
            for (String topic : result.unknownLeaderTopics)
                this.metadata.add(topic, now);
            ...
            // 針對(duì)這個(gè) topic 集合進(jìn)行元數(shù)據(jù)更新
            this.metadata.requestUpdate();
        }
        // 4. 循環(huán) readyNodes 并檢查客戶端與要發(fā)送節(jié)點(diǎn)的網(wǎng)絡(luò)是否已經(jīng)建立好了
        Iterator<Node> iter = result.readyNodes.iterator();
        long notReadyTimeout = Long.MAX_VALUE;
        while (iter.hasNext()) {
            Node node = iter.next();
            // 檢查客戶端與要發(fā)送節(jié)點(diǎn)的網(wǎng)絡(luò)是否已經(jīng)建立好了
            if (!this.client.ready(node, now)) {
                // 移除對(duì)應(yīng)節(jié)點(diǎn)
                iter.remove();
                notReadyTimeout = Math.min(notReadyTimeout, this.client.pollDelayMs(node, now));
            }
        }
        // create produce requests
        // 5. 獲取上面返回的已經(jīng)準(zhǔn)備好的節(jié)點(diǎn)上要發(fā)送的 ProducerBatch 集合
        Map<Integer, List<ProducerBatch>> batches = this.accumulator.drain(cluster, result.readyNodes, this.maxRequestSize, now);
        // 6. 將從消息累加器中讀取的數(shù)據(jù)集，放入正在執(zhí)行發(fā)送相關(guān)的消息批次集合中
        addToInflightBatches(batches);
        // 7. 要保證消息的順序性，即參數(shù) max.in.flight.requests.per.cnotallow=1
        if (guaranteeMessageOrder) {
            // Mute all the partitions drained
            for (List<ProducerBatch> batchList : batches.values()) {
                for (ProducerBatch batch : batchList)
                    // 對(duì) tp 進(jìn)行 mute，保證只有一個(gè) batch 正在發(fā)送
                    this.accumulator.mutePartition(batch.topicPartition);
            }
        }
        // 重置下一批次的過(guò)期時(shí)間
        accumulator.resetNextBatchExpiryTime();
        // 8. 從正在執(zhí)行發(fā)送數(shù)據(jù)集合 inflightBatches 中獲取過(guò)期集合
        List<ProducerBatch> expiredInflightBatches = getExpiredInflightBatches(now);
        // 9. 從 batches 中獲取過(guò)期集合
        List<ProducerBatch> expiredBatches = this.accumulator.expiredBatches(now);
        // 10. 從 inflightBatches 與 batches 中查找已過(guò)期的消息批次(ProducerBatch)，判斷批次是否過(guò)      期是根據(jù)系統(tǒng)當(dāng)前時(shí)間與 ProducerBatch 創(chuàng)建時(shí)間之差是否超過(guò)120s，過(guò)期時(shí)間可以通過(guò)參數(shù) delivery.timeout.ms 設(shè)置。
        expiredBatches.addAll(expiredInflightBatches);

        // 如果過(guò)期批次不為空 則輸出對(duì)應(yīng)日志
        if (!expiredBatches.isEmpty())
            log.trace("Expired {} batches in accumulator", expiredBatches.size());
        // 11. 處理已過(guò)期的消息批次，通知該批消息發(fā)送失敗并返回給客戶端
        for (ProducerBatch expiredBatch : expiredBatches) {
            // 處理當(dāng)前過(guò)期ProducerBatch的回調(diào)結(jié)果 ProduceRequestResult,并且設(shè)置超時(shí)異常 new TimeoutException(errorMessage)
            String errorMessage = "Expiring " + expiredBatch.recordCount + " record(s) for " + expiredBatch.topicPartition
                + ":" + (now - expiredBatch.createdMs) + " ms has passed since batch creation";
            // 通知該批消息發(fā)送失敗并返回給客戶端
            failBatch(expiredBatch, -1, NO_TIMESTAMP, new TimeoutException(errorMessage), false);
            // ... 事務(wù)管理器的處理忽略
        }
        // 收集統(tǒng)計(jì)指標(biāo)，后續(xù)會(huì)專門(mén)對(duì) Kafka 的 Metrics 進(jìn)行分析
        sensors.updateProduceRequestMetrics(batches);

        // 設(shè)置下一次的發(fā)送延時(shí)
        long pollTimeout = Math.min(result.nextReadyCheckDelayMs, notReadyTimeout);
        pollTimeout = Math.min(pollTimeout, this.accumulator.nextExpiryTimeMs() - now);
        pollTimeout = Math.max(pollTimeout, 0);
        if (!result.readyNodes.isEmpty()) {
            log.trace("Nodes with data ready to send: {}", result.readyNodes);
            pollTimeout = 0;
        }
        // 12. 發(fā)送消息暫存到 NetworkClient send 字段里。
        sendProduceRequests(batches, now);
        return pollTimeout;
}

void runOnce() {
        // 1. 把消息緩存在 KafkaChannel 的 Send 字段里。
        long pollTimeout = sendProducerData(currentTimeMs);
        // 2. 讀取消息實(shí)現(xiàn)真正的網(wǎng)絡(luò)發(fā)送
        client.poll(pollTimeout, currentTimeMs);
}