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)絡連接進行網(wǎng)絡讀寫
    /* the record accumulator that batches records */
    private final RecordAccumulator accumulator; // 消息倉庫累加器
    /* 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ā)送確認機制
    /* the number of times to retry a failed request before giving up */
    private final int retries; // 發(fā)送失敗后的重試次數(shù)，默認為0次

    /* true while the sender thread is still running */
    private volatile boolean running; // Sender 線程是否還在運行中     
    /* true when the caller wants to ignore all unsent/inflight messages and force close.  */
    private volatile boolean forceClose; // 是否強制關閉，此時會忽略正在發(fā)送中的消息。
    /* the max time to wait for the server to respond to the request*/
    private final int requestTimeoutMs; // 等待服務端響應的最大時間,默認30s       
    /* The max time to wait before retrying a request which has failed */
    private final long retryBackoffMs; // 失敗重試退避時間      
    /* 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; // 事務管理，這里忽略 后續(xù)會有專門一篇講解事務相關的
    // 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ā)送相關的消息批次集合， 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 線程是否在運行狀態(tài)中。
        // 1. 如果 Sender 線程在運行狀態(tài)即 running=true，則一直循環(huán)調用 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. 如果（沒有強制關閉 && ((消息累加器中還有剩余消息待發(fā)送 || 還有等待未響應的消息 ) || 還有事務請求未完成))，則繼續(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. 對進行中的事務進行中斷，則繼續(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. 如果強制關閉，則關閉事務管理器、終止消息的追加并清空未完成的批次
        if (forceClose) {
            if (transactionManager != null) {
                log.debug("Aborting incomplete transactional requests due to forced shutdown");
                // 關閉事務管理器
                transactionManager.close();
            }
            log.debug("Aborting incomplete batches due to forced shutdown");
            // 終止消息的追加并清空未完成的批次
            this.accumulator.abortIncompleteBatches();
        }
        // 5. 關閉 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) {
           ... //事務處理方法 后續(xù)文章專門講解 
        }
        // 1. 獲取當前時間的時間戳。
        long currentTimeMs = time.milliseconds();
        // 2. 調用 sendProducerData 發(fā)送消息,但并非真正的發(fā)送，而是把消息緩存在 把消息緩存在 KafkaChannel 的 Send 字段里。下一篇會講解 NetworkClient。
        long pollTimeout = sendProducerData(currentTimeMs);
        // 3. 讀取消息實現(xiàn)真正的網(wǎng)絡發(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)準備好的節(jié)點以及找不到 Leader 分區(qū)對應的節(jié)點的主題
        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ū)對應的節(jié)點不存在，則強制更新元數(shù)據(jù)
        if (!result.unknownLeaderTopics.isEmpty()) {
            // 添加 topic 到?jīng)]有拉取到元數(shù)據(jù)的 topic 集合中，并標識需要更新元數(shù)據(jù)
            for (String topic : result.unknownLeaderTopics)
                this.metadata.add(topic, now);
            ...
            // 針對這個 topic 集合進行元數(shù)據(jù)更新
            this.metadata.requestUpdate();
        }
        // 4. 循環(huán) readyNodes 并檢查客戶端與要發(fā)送節(jié)點的網(wǎng)絡是否已經(jīng)建立好了
        Iterator<Node> iter = result.readyNodes.iterator();
        long notReadyTimeout = Long.MAX_VALUE;
        while (iter.hasNext()) {
            Node node = iter.next();
            // 檢查客戶端與要發(fā)送節(jié)點的網(wǎng)絡是否已經(jīng)建立好了
            if (!this.client.ready(node, now)) {
                // 移除對應節(jié)點
                iter.remove();
                notReadyTimeout = Math.min(notReadyTimeout, this.client.pollDelayMs(node, now));
            }
        }
        // create produce requests
        // 5. 獲取上面返回的已經(jīng)準備好的節(jié)點上要發(fā)送的 ProducerBatch 集合
        Map<Integer, List<ProducerBatch>> batches = this.accumulator.drain(cluster, result.readyNodes, this.maxRequestSize, now);
        // 6. 將從消息累加器中讀取的數(shù)據(jù)集，放入正在執(zhí)行發(fā)送相關的消息批次集合中
        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)
                    // 對 tp 進行 mute，保證只有一個 batch 正在發(fā)送
                    this.accumulator.mutePartition(batch.topicPartition);
            }
        }
        // 重置下一批次的過期時間
        accumulator.resetNextBatchExpiryTime();
        // 8. 從正在執(zhí)行發(fā)送數(shù)據(jù)集合 inflightBatches 中獲取過期集合
        List<ProducerBatch> expiredInflightBatches = getExpiredInflightBatches(now);
        // 9. 從 batches 中獲取過期集合
        List<ProducerBatch> expiredBatches = this.accumulator.expiredBatches(now);
        // 10. 從 inflightBatches 與 batches 中查找已過期的消息批次(ProducerBatch)，判斷批次是否過      期是根據(jù)系統(tǒng)當前時間與 ProducerBatch 創(chuàng)建時間之差是否超過120s，過期時間可以通過參數(shù) delivery.timeout.ms 設置。
        expiredBatches.addAll(expiredInflightBatches);

        // 如果過期批次不為空 則輸出對應日志
        if (!expiredBatches.isEmpty())
            log.trace("Expired {} batches in accumulator", expiredBatches.size());
        // 11. 處理已過期的消息批次，通知該批消息發(fā)送失敗并返回給客戶端
        for (ProducerBatch expiredBatch : expiredBatches) {
            // 處理當前過期ProducerBatch的回調結果 ProduceRequestResult,并且設置超時異常 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);
            // ... 事務管理器的處理忽略
        }
        // 收集統(tǒng)計指標，后續(xù)會專門對 Kafka 的 Metrics 進行分析
        sensors.updateProduceRequestMetrics(batches);

        // 設置下一次的發(fā)送延時
        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. 讀取消息實現(xiàn)真正的網(wǎng)絡發(fā)送
        client.poll(pollTimeout, currentTimeMs);
}