Kafka 源码解析：生产者运行机制

Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-demo");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
KafkaProducer<Integer, String> producer = new KafkaProducer<>(properties);

boolean isAsync = true;

for (int i = 0; i < 10; i++) {
    if (isAsync) {
        // 异步发送消息
        producer.send(new ProducerRecord<>(DEFAULT_TOPIC, i, "zhenchao"), (metadata, e) -> {
            if (null != e) {
                // ... 处理错误
                return;
            }
            printResult(metadata); // 打印结果信息
        });
    } else {
        // 同步发送消息，指定 topic 和消息内容
        Future<RecordMetadata> future = producer.send(new ProducerRecord<>(DEFAULT_TOPIC, i, "zhenchao"));
        RecordMetadata metadata = future.get(10, TimeUnit.SECONDS);
        this.printResult(metadata); // 打印结果信息

    }
}

public class ProducerRecord<K, V> {

    /** 主题 */
    private final String topic;
    /** 分区 */
    private final Integer partition;
    /** 消息对应的 key */
    private final K key;
    /** 消息内容 */
    private final V value;
    /** 时间戳 */
    private final Long timestamp;

    // ... 省略方法定义

}

public class KafkaProducer<K, V> implements Producer<K, V> {

    /** clientId 生成器，如果没有明确指定客户端 ID，则使用该字段顺序生成一个 */
    private static final AtomicInteger PRODUCER_CLIENT_ID_SEQUENCE = new AtomicInteger(1);
    /** 生产者唯一标识（对应 client.id 属性配置 ） */
    private String clientId;
    /** 分区选择器（对应 partitioner.class 属性配置），如果未明确指定分区，则基于一定的策略为消息选择合适的分区 */
    private final Partitioner partitioner;
    /** 消息的最大长度（对应 max.request.size 配置，包含消息头、序列化之后的 key 和 value） */
    private final int maxRequestSize;
    /** 发送单条消息的缓冲区大小（对应 buffer.memory 配置） */
    private final long totalMemorySize;
    /** kafka 集群元数据 */
    private final Metadata metadata;
    /** 消息收集器，用于收集并缓存消息，等待 Sender 线程的发送 */
    private final RecordAccumulator accumulator;
    /** 消息发送线程对象 */
    private final Sender sender;
    /** 消息发送线程 */
    private final Thread ioThread;
    /** 压缩算法（对应 compression.type 配置） */
    private final CompressionType compressionType;
    /** 时间戳工具 */
    private final Time time;
    /** key 序列化器（对应 key.serializer 配置） */
    private final Serializer<K> keySerializer;
    /** value 序列化器（对应 value.serializer 配置） */
    private final Serializer<V> valueSerializer;
    /** 封装配置信息 */
    private final ProducerConfig producerConfig;
    /** 等待更新 kafka 集群元数据的最大时长 */
    private final long maxBlockTimeMs;
    /** 消息发送的超时时间（从发送到收到 ACK 响应） */
    private final int requestTimeoutMs;
    /** 发送拦截器（对应 interceptor.classes 配置），用于待发送的消息进行拦截并修改，也可以对 ACK 响应进行拦截处理 */
    private final ProducerInterceptors<K, V> interceptors;

    // ... 省略方法定义

}

private KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
    try {
        log.trace("Starting the Kafka producer");
        // 获取用户配置信息
        Map<String, Object> userProvidedConfigs = config.originals();
        this.producerConfig = config;
        this.time = Time.SYSTEM;

        // 尝试获取用户配置的 clientId，如果未配置则基于 PRODUCER_CLIENT_ID_SEQUENCE 顺序生成一个
        this.clientId = config.getString(ProducerConfig.CLIENT_ID_CONFIG);
        if (clientId.length() <= 0) {
            // 用户未指定 clientId，基于 PRODUCER_CLIENT_ID_SEQUENCE 顺序生成一个
            clientId = "producer-" + PRODUCER_CLIENT_ID_SEQUENCE.getAndIncrement();
        }

        // ... 省略打点相关注册逻辑

        // 获取配置的分区器对象（反射创建）
        this.partitioner = config.getConfiguredInstance(ProducerConfig.PARTITIONER_CLASS_CONFIG, Partitioner.class);
        // 获取生产者重试间隔
        long retryBackoffMs = config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG);

        // 如果参数未指定 key 序列化器，则尝试从配置中获取 key 序列化器对象（反射创建）
        if (keySerializer == null) {
            this.keySerializer = config.getConfiguredInstance(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, Serializer.class);
            this.keySerializer.configure(config.originals(), true);
        } else {
            config.ignore(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG);
            this.keySerializer = keySerializer;
        }

        // 如果参数未指定 value 序列化器，则尝试从配置中获取 value 序列化器对象（反射创建）
        if (valueSerializer == null) {
            this.valueSerializer = config.getConfiguredInstance(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, Serializer.class);
            this.valueSerializer.configure(config.originals(), false);
        } else {
            config.ignore(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG);
            this.valueSerializer = valueSerializer;
        }

        // load interceptors and make sure they get clientId
        userProvidedConfigs.put(ProducerConfig.CLIENT_ID_CONFIG, clientId);

        // 获取注册的拦截器列表
        List<ProducerInterceptor<K, V>> interceptorList = (List) (new ProducerConfig(userProvidedConfigs, false))
                .getConfiguredInstances(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptor.class);
        this.interceptors = interceptorList.isEmpty() ? null : new ProducerInterceptors<>(interceptorList);

        ClusterResourceListeners clusterResourceListeners =
                this.configureClusterResourceListeners(keySerializer, valueSerializer, interceptorList, reporters);

        // 创建并更新 kafka 集群的元数据信息
        this.metadata = new Metadata(retryBackoffMs, config.getLong(ProducerConfig.METADATA_MAX_AGE_CONFIG), true, clusterResourceListeners);
        // 获取并设置生产者发送请求的大小
        this.maxRequestSize = config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG);
        // 获取并设置生产者内存缓冲区大小，用于缓存要发送到服务器的消息
        this.totalMemorySize = config.getLong(ProducerConfig.BUFFER_MEMORY_CONFIG);
        // 获取并设置消息压缩算法，可以设置为 snappy、gzip 或 lz4，默认不压缩。
        this.compressionType = CompressionType.forName(config.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));

        // ... 基于用户配置设置 maxBlockTimeMs 和 requestTimeoutMs，省略

        // 创建消息收集器，用于异步发送消息
        this.accumulator = new RecordAccumulator(
                config.getInt(ProducerConfig.BATCH_SIZE_CONFIG), // 指定每个批次的大小（单位：字节）
                this.totalMemorySize,
                this.compressionType,
                config.getLong(ProducerConfig.LINGER_MS_CONFIG), // 消息缓存超时发送时间
                retryBackoffMs,
                metrics,
                time);

        // 获取 kafka 集群主机列表
        List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(config.getList(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
        // 更新 kafka 集群元数据信息
        this.metadata.update(Cluster.bootstrap(addresses), Collections.<String>emptySet(), time.milliseconds());
        ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config.values());
        // 创建 NetworkClient 对象，NetworkClient 是 producer 网络 I/O 的核心
        NetworkClient client = new NetworkClient(
                new Selector(config.getLong(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), metrics, time, "producer", channelBuilder),
                metadata,
                clientId,
                config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION),
                config.getLong(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG),
                config.getInt(ProducerConfig.SEND_BUFFER_CONFIG),
                config.getInt(ProducerConfig.RECEIVE_BUFFER_CONFIG),
                requestTimeoutMs,
                time,
                true);

        // 创建并启动 Sender 线程
        this.sender = new Sender(
                client,
                metadata,
                accumulator,
                config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION) == 1,
                config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
                (short) parseAcks(config.getString(ProducerConfig.ACKS_CONFIG)),
                config.getInt(ProducerConfig.RETRIES_CONFIG),
                metrics,
                Time.SYSTEM,
                requestTimeoutMs);
        String ioThreadName = "kafka-producer-network-thread" + (clientId.length() > 0 ? " | " + clientId : "");
        this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
        this.ioThread.start();

        // 打印未使用的配置
        config.logUnused();
        AppInfoParser.registerAppInfo(JMX_PREFIX, clientId);
        log.debug("Kafka producer started");
    } catch (Throwable t) {
        // ... 省略异常处理
    }
}

public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    // 遍历注册的拦截器对待发送的消息执行拦截修改
    ProducerRecord<K, V> interceptedRecord = this.interceptors == null ? record : this.interceptors.onSend(record);
    // 调用 doSend 方法开始发送消息
    return this.doSend(interceptedRecord, callback);
}

private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
    TopicPartition tp = null;
    try {
        // 1. 获取 kafka 集群元数据信息，如果当前请求的是新 topic，或者指定的分区超过已知的分区范围，则会触发更新集群元数据信息
        ClusterAndWaitTime clusterAndWaitTime = this.waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs);
        long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);
        Cluster cluster = clusterAndWaitTime.cluster;

        // 2 基于注册的序列化器对 key 执行序列化
        byte[] serializedKey;
        try {
            serializedKey = keySerializer.serialize(record.topic(), 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");
        }

        // 3. 基于注册的序列化器对 value 执行序列化
        byte[] serializedValue;
        try {
            serializedValue = valueSerializer.serialize(record.topic(), 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");
        }

        // 4. 为当前消息选择合适的分区，如果未明确指定的话，则基于注册的分区器为当前消息计算分区
        int partition = this.partition(record, serializedKey, serializedValue, cluster);

        /* 5. 将消息追加到消息收集器（RecordAccumulator）中 */

        // 计算当前消息大小，并校验消息是否过大
        int serializedSize = Records.LOG_OVERHEAD + Record.recordSize(serializedKey, serializedValue);
        this.ensureValidRecordSize(serializedSize);
        tp = new TopicPartition(record.topic(), partition); // 消息投递的目标 topic 分区
        // 如果未明确为当前消息指定时间戳，则设置为当前时间戳
        long timestamp = record.timestamp() == null ? time.milliseconds() : record.timestamp();
        log.trace("Sending record {} with callback {} to topic {} partition {}", record, callback, record.topic(), partition);
        // producer callback will make sure to call both 'callback' and interceptor callback
        Callback interceptCallback = this.interceptors == null ? callback : new InterceptorCallback<>(callback, this.interceptors, tp);
        // 追加消息到收集器中
        RecordAccumulator.RecordAppendResult result = accumulator.append(
                tp, timestamp, serializedKey, serializedValue, interceptCallback, remainingWaitMs);

        /* 6. 条件性唤醒消息发送线程 */

        // 如果队列中不止一个 RecordBatch，或者最后一个 RecordBatch 满了，或者有创建新的 RecordBatch，则唤醒 Sender 线程发送消息
        if (result.batchIsFull || result.newBatchCreated) {
            log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), partition);
            // 唤醒 sender 线程，发送消息
            this.sender.wakeup();
        }
        return result.future;
    }
    // ... 省略异常处理
}

public final class Metadata {

    /** 元数据最小更新时间间隔，默认是 100 毫秒，防止更新太频繁 */
    private final long refreshBackoffMs;
    /** 元数据更新时间间隔，默认为 5 分钟 */
    private final long metadataExpireMs;
    /** 元数据版本号，每更新成功一次则版本号加 1 */
    private int version;
    /** 上一次更新元数据的时间戳，不管成功还是失败 */
    private long lastRefreshMs;
    /** 上一次成功更新元数据的时间戳 */
    private long lastSuccessfulRefreshMs;
    /** 集群信息 */
    private Cluster cluster;
    /** 标记是否需要更新集群元数据信息 */
    private boolean needUpdate;
    /** 记录集群中所有的 topic 信息，key 是 topic，value 是 topic 过期的时间戳 */
    private final Map<String, Long> topics;
    /** 元数据更新监听器 */
    private final List<Listener> listeners;
    /** 标记是否需要更新所有 topic 的元数据信息，一般只更新当前用到的 topic 的元数据信息 */
    private boolean needMetadataForAllTopics;
    /** 是否允许 topic 过期 */
    private final boolean topicExpiryEnabled;

    // ... 省略方法定义

}

private ClusterAndWaitTime waitOnMetadata(String topic, Integer partition, long maxWaitMs) throws InterruptedException {
    // 添加 topic 到集合中，如果是新 topic，标记需要更新集群元数据信息
    metadata.add(topic);
    // 获取当前集群信息
    Cluster cluster = metadata.fetch();
    // 获取指定 topic 的分区数目
    Integer partitionsCount = cluster.partitionCountForTopic(topic);

    // 如果参数未指定分区，或指定的分区在当前记录的分区范围之内，则返回历史集群信息
    if (partitionsCount != null && (partition == null || partition < partitionsCount)) {
        return new ClusterAndWaitTime(cluster, 0);
    }

    /* 否则，当前缓存的集群元数据信息可能已经过期，需要进行更新 */

    long begin = time.milliseconds();
    long remainingWaitMs = maxWaitMs; // 剩余等待时间
    long elapsed;

    /* 请求集群的元数据信息，直到获取到信息或者超时 */
    do {
        log.trace("Requesting metadata update for topic {}.", topic);
        // 更新 Metadata 的 needUpdate 字段，并获取当前元数据的版本号
        int version = metadata.requestUpdate();
        // 唤醒 sender 线程，由 sender 线程负责更新元数据信息
        sender.wakeup();
        try {
            // 等待元数据更新完成
            metadata.awaitUpdate(version, remainingWaitMs);
        } catch (TimeoutException ex) {
            // 等待超时
            throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
        }

        // 获取更新后的集群信息
        cluster = metadata.fetch();
        elapsed = time.milliseconds() - begin;
        if (elapsed >= maxWaitMs) {
            // 等待超时
            throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
        }
        // 权限检测
        if (cluster.unauthorizedTopics().contains(topic)) {
            throw new TopicAuthorizationException(topic);
        }
        remainingWaitMs = maxWaitMs - elapsed; // 更新剩余等待时间
        partitionsCount = cluster.partitionCountForTopic(topic); // 获取指定 topic 的分区数目
    } while (partitionsCount == null); // 更新集群信息失败，继续重试

    /* 更新集群信息成功 */

    // 参数指定的分区非法
    if (partition != null && partition >= partitionsCount) {
        throw new KafkaException(String.format("Invalid partition given with record: %d is not in the range [0...%d).", partition, partitionsCount));
    }

    return new ClusterAndWaitTime(cluster, elapsed);
}

public final class Cluster {

    /** kafka 集群中的节点信息列表（包括 id、host、port 等信息） */
    private final List<Node> nodes;
    /** 未授权的 topic 集合 */
    private final Set<String> unauthorizedTopics;
    /** 内部 topic 集合 */
    private final Set<String> internalTopics;
    /** 记录 topic 分区与分区详细信息的映射关系 */
    private final Map<TopicPartition, PartitionInfo> partitionsByTopicPartition;
    /** 记录 topic 及其分区信息的映射关系 */
    private final Map<String, List<PartitionInfo>> partitionsByTopic;
    /** 记录 topic 及其分区信息的映射关系（必须包含 leader 副本） */
    private final Map<String, List<PartitionInfo>> availablePartitionsByTopic;
    /** 记录节点 ID 与分区信息的映射关系 */
    private final Map<Integer, List<PartitionInfo>> partitionsByNode;
    /** key 是 brokerId，value 是 broker 节点信息，方便基于 brokerId 获取对应的节点信息 */
    private final Map<Integer, Node> nodesById;

    // ... 省略方法定义

}

private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
    // 获取当前待发送消息所指定的分区
    Integer partition = record.partition();
    // 如果未指定分区，则为当前消息计算一个分区编号
    return partition != null ?
            partition :
            partitioner.partition(record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
}

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    // 获取当前 topic 的分区详细信息
    List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
    // 获取当前 topic 对应的分区数
    int numPartitions = partitions.size();
    // 如果没有设置 key，则基于轮询算法
    if (keyBytes == null) {
        // 获取当前 topic 对应的上次位置值加 1，如果是第一次则随机生成一个
        int nextValue = this.nextValue(topic);
        // 获取当前 topic 包含 leader 副本的分区详细信息
        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;
        }
    }
    // 如果指定了 key，则使用 murmur2 算法对 key 做哈希取模
    else {
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }
}

public final class RecordAccumulator {

    /** 标识当前收集器是否被关闭，对应 producer 被关闭 */
    private volatile boolean closed;
    /** 记录正在执行 flush 操作的线程数 */
    private final AtomicInteger flushesInProgress;
    /** 记录正在执行 append 操作的线程数 */
    private final AtomicInteger appendsInProgress;
    /** 指定每个 RecordBatch 中 ByteBuffer 的大小 */
    private final int batchSize;
    /** 消息压缩类型 */
    private final CompressionType compression;
    /** 通过参数 linger.ms 指定，当本地消息缓存时间超过该值时，即使消息量未达到阈值也会进行投递 */
    private final long lingerMs;
    /** 生产者重试时间间隔 */
    private final long retryBackoffMs;
    /** 缓存（ByteBuffer）管理工具 */
    private final BufferPool free;
    /** 时间戳工具 */
    private final Time time;
    /** 记录 topic 分区与 RecordBatch 的映射关系，对应的消息都是发往对应的 topic 分区 */
    private final ConcurrentMap<TopicPartition, Deque<RecordBatch>> batches;
    /** 记录未发送完成（即未收到服务端响应）的消息集合 */
    private final IncompleteRecordBatches incomplete;
    /**
     * 消息顺序性保证，
     * 缓存当前待发送消息的目标 topic 分区，防止对于同一个 topic 分区同时存在多个未完成的消息，可能导致消息顺序性错乱
     */
    private final Set<TopicPartition> muted;
    /** 记录 drain 方法批量导出消息时上次的偏移量 */
    private int drainIndex;

    // ... 省略方法定义

}

public final class RecordBatch {

    /** 当前 RecordBatch 创建的时间戳 */
    final long createdMs;
    /** 当前缓存的消息的目标 topic 分区 */
    final TopicPartition topicPartition;
    /** 标识当前 RecordBatch 发送之后的状态 */
    final ProduceRequestResult produceFuture;
    /** 消息的 Callback 队列，每个消息都对应一个 Callback 对象 */
    private final List<Thunk> thunks = new ArrayList<>();
    /** 用来存储数据的 {@link MemoryRecords} 对应的 builder 对象 */
    private final MemoryRecordsBuilder recordsBuilder;
    /** 发送当前 RecordBatch 的重试次数 */
    volatile int attempts;
    /** 最后一次重试发送的时间戳` */
    long lastAttemptMs;
    /** 记录保存的 record 个数 */
    int recordCount;
    /** 记录最大的 record 字节数 */
    int maxRecordSize;
    /** 记录上次投递当前 BatchRecord 的时间戳 */
    long drainedMs;
    /** 追后一次向当前 RecordBatch 追加消息的时间戳 */
    long lastAppendTime;
    /** 标记是否正在重试 */
    private boolean retry;

    // ... 省略方法定义

}

public RecordAppendResult append(TopicPartition tp,
                                 long timestamp,
                                 byte[] key,
                                 byte[] value,
                                 Callback callback,
                                 long maxTimeToBlock) throws InterruptedException {
    // 记录正在向收集器中追加消息的线程数
    appendsInProgress.incrementAndGet();
    try {
        // 获取当前 topic 分区对应的 Deque，如果不存在则创建一个
        Deque<RecordBatch> dq = this.getOrCreateDeque(tp);
        synchronized (dq) {
            if (closed) {
                // producer 已经被关闭了，抛出异常
                throw new IllegalStateException("Cannot send after the producer is closed.");
            }
            // 向 Deque 中最后一个 RecordBatch 追加 Record，并返回对应的 RecordAppendResult 对象
            RecordAppendResult appendResult = this.tryAppend(timestamp, key, value, callback, dq);
            if (appendResult != null) {
                // 追加成功，直接返回
                return appendResult;
            }
        }

        /* 追加 Record 失败，尝试申请新的 buffer */

        int size = Math.max(this.batchSize, Records.LOG_OVERHEAD + Record.recordSize(key, value));
        log.trace("Allocating a new {} byte message buffer for topic {} partition {}", size, tp.topic(), tp.partition());
        // 申请新的 buffer
        ByteBuffer buffer = free.allocate(size, maxTimeToBlock);
        synchronized (dq) {
            if (closed) {
                // 再次校验 producer 状态，如果已经被关闭了，抛出异常
                throw new IllegalStateException("Cannot send after the producer is closed.");
            }

            // 再次尝试向 Deque 中最后一个 RecordBatch 追加 Record
            RecordAppendResult appendResult = this.tryAppend(timestamp, key, value, callback, dq);
            if (appendResult != null) {
                // 追加成功则返回，同时归还之前申请的 buffer
                free.deallocate(buffer);
                return appendResult;
            }

            /* 仍然追加失败，创建一个新的 RecordBatch 进行追加 */

            MemoryRecordsBuilder recordsBuilder = MemoryRecords.builder(buffer, compression, TimestampType.CREATE_TIME, this.batchSize);
            RecordBatch batch = new RecordBatch(tp, recordsBuilder, time.milliseconds());
            // 在新创建的 RecordBatch 中追加 Record
            FutureRecordMetadata future = Utils.notNull(batch.tryAppend(timestamp, key, value, callback, time.milliseconds()));
            dq.addLast(batch);
            // 追加到未完成的集合中
            incomplete.add(batch);
            // 封装成 RecordAppendResult 对象返回
            return new RecordAppendResult(future, dq.size() > 1 || batch.isFull(), true);
        }
    } finally {
        appendsInProgress.decrementAndGet();
    }
}

private RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, Callback callback, Deque<RecordBatch> deque) {
    // 获取 deque 的最后一个 RecordBatch
    RecordBatch last = deque.peekLast();
    if (last != null) {
        // 尝试往该 RecordBatch 末尾追加消息
        FutureRecordMetadata future = last.tryAppend(timestamp, key, value, callback, time.milliseconds());
        if (future == null) {
            // 追加失败
            last.close();
        } else {
            // 追加成功，将结果封装成 RecordAppendResult 对象返回
            return new RecordAppendResult(future, deque.size() > 1 || last.isFull(), false);
        }
    }
    return null;
}

// org.apache.kafka.clients.producer.internals.RecordBatch#tryAppend
public FutureRecordMetadata tryAppend(long timestamp, byte[] key, byte[] value, Callback callback, long now) {
    // 检测是否还有多余的空间容纳该消息
    if (!recordsBuilder.hasRoomFor(key, value)) {
        // 没有多余的空间则直接返回，后面会尝试申请新的空间
        return null;
    }
    // 添加当前消息到 MemoryRecords，并返回消息对应的 CRC32 校验码
    long checksum = this.recordsBuilder.append(timestamp, key, value);
    // 更新最大 record 字节数
    this.maxRecordSize = Math.max(this.maxRecordSize, Record.recordSize(key, value));
    // 更新最后一次追加记录时间戳
    this.lastAppendTime = now;
    FutureRecordMetadata future = new FutureRecordMetadata(
            produceFuture, recordCount,
            timestamp, checksum,
            key == null ? -1 : key.length,
            value == null ? -1 : value.length);
    if (callback != null) {
        // 如果指定了 Callback，将 Callback 和 FutureRecordMetadata 封装到 Trunk 中
        thunks.add(new Thunk(callback, future));
    }
    this.recordCount++;
    return future;
}

public void run() {

    // 主循环，一直运行直到 KafkaProducer 被关闭
    while (running) {
        try {
            this.run(time.milliseconds());
        } catch (Exception e) {
            log.error("Uncaught error in kafka producer I/O thread: ", e);
        }
    }

    /* 如果 KafkaProducer 被关闭，尝试发送剩余的消息 */
    while (!forceClose // 不是强制关闭
            // 存在未发送或已发送待响应的请求
            && (this.accumulator.hasUnsent() || this.client.inFlightRequestCount() > 0)) {
        try {
            this.run(time.milliseconds());
        } catch (Exception e) {
            log.error("Uncaught error in kafka producer I/O thread: ", e);
        }
    }

    // 如果是强制关闭，忽略所有未发送和已发送待响应的请求
    if (forceClose) {
        // 丢弃所有未发送完成的消息
        this.accumulator.abortIncompleteBatches();
    }
    try {
        // 关闭网络连接
        this.client.close();
    } catch (Exception e) {
        log.error("Failed to close network client", e);
    }
}

void run(long now) {

    // 1. 计算需要以及可以向哪些节点发送请求
    Cluster cluster = metadata.fetch(); // 获取 kafka 集群信息
    RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now); // 计算需要向哪些节点发送请求

    // 2. 如果存在未知的 leader 副本对应的节点（对应的 topic 分区正在执行 leader 选举，或者对应的 topic 已经失效），标记需要更新缓存的集群元数据信息
    if (!result.unknownLeaderTopics.isEmpty()) {
        for (String topic : result.unknownLeaderTopics) this.metadata.add(topic);
        this.metadata.requestUpdate();
    }

    // 3. 遍历处理待发送请求的目标节点，基于网络 IO 检查对应节点是否可用，对于不可用的节点则剔除
    Iterator<Node> iter = result.readyNodes.iterator();
    long notReadyTimeout = Long.MAX_VALUE;
    while (iter.hasNext()) {
        Node node = iter.next();
        // 检查目标节点是否准备好接收请求，如果未准备好但目标节点允许创建连接，则创建到目标节点的连接
        if (!this.client.ready(node, now)) {
            // 对于未准备好的节点，则从 ready 集合中删除
            iter.remove();
            notReadyTimeout = Math.min(notReadyTimeout, this.client.connectionDelay(node, now));
        }
    }

    // 4. 获取每个节点待发送消息集合，其中 key 是目标 leader 副本所在节点 ID
    Map<Integer, List<RecordBatch>> batches =
            this.accumulator.drain(cluster, result.readyNodes, this.maxRequestSize, now);

    // 5. 如果需要保证消息的强顺序性，则缓存对应 topic 分区对象，防止同一时间往同一个 topic 分区发送多条处于未完成状态的消息
    if (guaranteeMessageOrder) {
        // 将所有 RecordBatch 的 topic 分区对象加入到 muted 集合中
        // 防止同一时间往同一个 topic 分区发送多条处于未完成状态的消息
        for (List<RecordBatch> batchList : batches.values()) {
            for (RecordBatch batch : batchList)
                this.accumulator.mutePartition(batch.topicPartition);
        }
    }

    // 6. 处理本地过期的消息，返回 TimeoutException，并释放空间
    List<RecordBatch> expiredBatches = this.accumulator.abortExpiredBatches(this.requestTimeout, now);

    // 如果存在待发送的消息，则设置 pollTimeout 等于 0，这样可以立即发送请求，从而能够缩短剩余消息的缓存时间，避免堆积
    long pollTimeout = Math.min(result.nextReadyCheckDelayMs, notReadyTimeout);
    if (!result.readyNodes.isEmpty()) {
        log.trace("Nodes with data ready to send: {}", result.readyNodes);
        pollTimeout = 0;
    }

    // 7. 发送请求到服务端，并处理服务端响应
    this.sendProduceRequests(batches, now);
    this.client.poll(pollTimeout, now);
}

public ReadyCheckResult ready(Cluster cluster, long nowMs) {
    // 用于记录接收请求的节点
    Set<Node> readyNodes = new HashSet<>();
    // 记录下次执行 ready 判断的时间间隔
    long nextReadyCheckDelayMs = Long.MAX_VALUE;
    // 记录找不到 leader 副本的分区对应的 topic 集合
    Set<String> unknownLeaderTopics = new HashSet<>();

    // 是否有线程在等待 BufferPool 分配空间
    boolean exhausted = this.free.queued() > 0;
    // 遍历每个 topic 分区及其 RecordBatch 队列，对每个分区的 leader 副本所在的节点执行判定
    for (Map.Entry<TopicPartition, Deque<RecordBatch>> entry : this.batches.entrySet()) {
        TopicPartition part = entry.getKey();
        Deque<RecordBatch> deque = entry.getValue();

        // 获取当前 topic 分区 leader 副本所在的节点
        Node leader = cluster.leaderFor(part);
        synchronized (deque) {
            // 当前分区 leader 副本未知，但存在发往该分区的消息
            if (leader == null && !deque.isEmpty()) {
                unknownLeaderTopics.add(part.topic());
            }
            // 如果需要保证消息顺序性，则不应该存在多个发往该 leader 副本节点且未完成的消息
            else if (!readyNodes.contains(leader) && !muted.contains(part)) {
                RecordBatch batch = deque.peekFirst();
                if (batch != null) {
                    // 当前为重试操作，且重试时间间隔未达到阈值时间
                    boolean backingOff = batch.attempts > 0 && batch.lastAttemptMs + retryBackoffMs > nowMs;
                    long waitedTimeMs = nowMs - batch.lastAttemptMs; // 重试等待的时间
                    long timeToWaitMs = backingOff ? retryBackoffMs : lingerMs;
                    long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);
                    boolean full = deque.size() > 1 || batch.isFull();
                    boolean expired = waitedTimeMs >= timeToWaitMs;

                    // 标记当前节点是否可以接收请求
                    boolean sendable = full // 1. 队列中有多个 RecordBatch，或第一个 RecordBatch 已满
                            || expired // 2. 当前等待重试的时间过长
                            || exhausted // 3. 有其他线程在等待 BufferPoll 分配空间，即本地消息缓存已满
                            || closed // 4. producer 已经关闭
                            || flushInProgress(); // 5. 有线程正在等待 flush 操作完成
                    if (sendable && !backingOff) {
                        // 允许发送消息，且当前为首次发送，或者重试等待时间已经较长，则记录目标 leader 副本所在节点
                        readyNodes.add(leader);
                    } else {
                        // 更新下次执行 ready 判定的时间间隔
                        nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);
                    }
                }
            }
        }
    }

    // 封装结果返回
    return new ReadyCheckResult(readyNodes, nextReadyCheckDelayMs, unknownLeaderTopics);
}

public Map<Integer, List<RecordBatch>> drain(Cluster cluster, Set<Node> nodes, int maxSize, long now) {
    if (nodes.isEmpty()) {
        return Collections.emptyMap();
    }

    // 记录转换后的结果，key 是目标节点 ID
    Map<Integer, List<RecordBatch>> batches = new HashMap<>();
    for (Node node : nodes) {
        int size = 0;
        // 获取当前节点上的分区信息
        List<PartitionInfo> parts = cluster.partitionsForNode(node.id());
        // 记录待发往当前节点的 RecordBatch 集合
        List<RecordBatch> ready = new ArrayList<>();
        /*
         * drainIndex 用于记录上次发送停止的位置，本次继续从当前位置开始发送，
         * 如果每次都是从 0 位置开始，可能会导致排在后面的分区饿死，可以看做是一个简单的负载均衡策略
         */
        int start = drainIndex = drainIndex % parts.size();
        do {
            PartitionInfo part = parts.get(drainIndex);
            TopicPartition tp = new TopicPartition(part.topic(), part.partition());
            // 如果需要保证消息强顺序性，则不应该同时存在多个发往目标分区的消息
            if (!muted.contains(tp)) {
                // 获取当前分区对应的 RecordBatch 集合
                Deque<RecordBatch> deque = this.getDeque(new TopicPartition(part.topic(), part.partition()));
                if (deque != null) {
                    synchronized (deque) {
                        RecordBatch first = deque.peekFirst();
                        if (first != null) {
                            // 重试 && 重试时间间隔未达到阈值时间
                            boolean backoff = first.attempts > 0 && first.lastAttemptMs + retryBackoffMs > now;
                            // 仅发送第一次发送，或重试等待时间较长的消息
                            if (!backoff) {
                                if (size + first.sizeInBytes() > maxSize && !ready.isEmpty()) {
                                    // 单次消息数据量已达到上限，结束循环，一般对应一个请求的大小，防止请求消息过大
                                    break;
                                } else {
                                    // 每次仅获取第一个 RecordBatch，并放入 read 列表中，这样给每个分区一个机会，保证公平，防止饥饿
                                    RecordBatch batch = deque.pollFirst();
                                    // 将当前 RecordBatch 设置为只读
                                    batch.close();
                                    size += batch.sizeInBytes();
                                    ready.add(batch);
                                    batch.drainedMs = now;
                                }
                            }
                        }
                    }
                }
            }
            // 更新 drainIndex
            this.drainIndex = (this.drainIndex + 1) % parts.size();
        } while (start != drainIndex);
        batches.put(node.id(), ready);
    }
    return batches;
}

private void sendProduceRequests(Map<Integer, List<RecordBatch>> collated, long now) {
    // 遍历处理待发送消息集合，key 是目标节点 ID
    for (Map.Entry<Integer, List<RecordBatch>> entry : collated.entrySet())
        this.sendProduceRequest(now, entry.getKey(), acks, requestTimeout, entry.getValue());
}

private void sendProduceRequest(long now, int destination, short acks, int timeout, List<RecordBatch> batches) {
    // 遍历 RecordBatch 集合，整理成 produceRecordsByPartition 和 recordsByPartition
    Map<TopicPartition, MemoryRecords> produceRecordsByPartition = new HashMap<>(batches.size());
    final Map<TopicPartition, RecordBatch> recordsByPartition = new HashMap<>(batches.size());
    for (RecordBatch batch : batches) {
        TopicPartition tp = batch.topicPartition;
        produceRecordsByPartition.put(tp, batch.records());
        recordsByPartition.put(tp, batch);
    }

    // 创建 ProduceRequest 请求构造器
    ProduceRequest.Builder requestBuilder = new ProduceRequest.Builder(acks, timeout, produceRecordsByPartition);

    // 创建回调对象，用于处理响应
    RequestCompletionHandler callback = new RequestCompletionHandler() {
        @Override
        public void onComplete(ClientResponse response) {
            handleProduceResponse(response, recordsByPartition, time.milliseconds());
        }
    };

    String nodeId = Integer.toString(destination);

    // 创建 ClientRequest 请求对象，如果 acks 不等于 0 则表示期望获取服务端响应
    ClientRequest clientRequest = client.newClientRequest(nodeId, requestBuilder, now, acks != 0, callback);
    // 缓存 ClientRequest 请求对象到 InFlightRequests 中
    client.send(clientRequest, now);
    log.trace("Sent produce request to {}: {}", nodeId, requestBuilder);
}

public List<ClientResponse> poll(long timeout, long now) {
    /*
     * 如果距离上次更新超过指定时间，且存在负载小的目标节点，
     * 则创建 MetadataRequest 请求更新本地缓存的集群元数据信息，并在下次执行 poll 操作时一并送出
     */
    long metadataTimeout = metadataUpdater.maybeUpdate(now);

    /* 发送网络请求 */
    try {
        this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
    } catch (IOException e) {
        log.error("Unexpected error during I/O", e);
    }

    /* 处理服务端响应 */

    long updatedNow = this.time.milliseconds();
    List<ClientResponse> responses = new ArrayList<>(); // 响应队列
    // 添加需要被丢弃的请求对应的响应到 responses 队列中，都是一些版本不匹配的请求
    this.handleAbortedSends(responses);
    // 对于发送成功且不期望服务端响应的请求，创建本地的响应对象添加到 responses 队列中
    this.handleCompletedSends(responses, updatedNow);
    /*
     * 获取并解析服务端响应
     * - 如果是更新集群元数据对应的响应，则更新本地缓存的集群元数据信息
     * - 如果是更新 API 版本的响应，则更新本地缓存的目标节点支持的 API 版本信息
     * - 否则，获取 ClientResponse 添加到 responses 队列中
     */
    this.handleCompletedReceives(responses, updatedNow);
    // 处理连接断开的请求，构建对应的 ClientResponse 添加到 responses 列表中，并标记需要更新集群元数据信息
    this.handleDisconnections(responses, updatedNow);
    // 处理 connections 列表，更新相应节点的连接状态
    this.handleConnections();
    // 如果需要更新本地的 API 版本信息，则创建对应的 ApiVersionsRequest 请求，并在下次执行 poll 操作时一并送出
    this.handleInitiateApiVersionRequests(updatedNow);
    // 遍历获取 inFlightRequests 中的超时请求，构建对应的 ClientResponse 添加到 responses 列表中，并标记需要更新集群元数据信息
    this.handleTimedOutRequests(responses, updatedNow);

    // 遍历处理响应对应的 onComplete 方法
    for (ClientResponse response : responses) {
        try {
            // 本质上就是在调用注册的 RequestCompletionHandler#onComplete 方法
            response.onComplete();
        } catch (Exception e) {
            log.error("Uncaught error in request completion:", e);
        }
    }

    return responses;
}

public long maybeUpdate(long now) {
    // 获取下次更新集群信息的时间戳
    long timeToNextMetadataUpdate = metadata.timeToNextUpdate(now);
    // 检查是否已经发送了 MetadataRequest 请求
    long waitForMetadataFetch = this.metadataFetchInProgress ? requestTimeoutMs : 0;
    // 计算当前距离下次发送 MetadataRequest 请求的时间差
    long metadataTimeout = Math.max(timeToNextMetadataUpdate, waitForMetadataFetch);
    if (metadataTimeout > 0) {
        // 如果时间还未到，则暂时不更新
        return metadataTimeout;
    }

    // 寻找负载最小的可用节点，如果没有可用的节点则返回 null
    Node node = leastLoadedNode(now);
    if (node == null) {
        log.debug("Give up sending metadata request since no node is available");
        return reconnectBackoffMs;
    }

    // 检查是否允许向目标节点发送请求，如果允许则创建 MetadataRequest 请求，并在下次执行 poll 操作时一并送出
    return this.maybeUpdate(now, node);
}

private long maybeUpdate(long now, Node node) {
    String nodeConnectionId = node.idString();

    // 如果允许向该节点发送请求
    if (canSendRequest(nodeConnectionId)) {
        // 标识正在请求更新集群元数据信息
        this.metadataFetchInProgress = true;
        // 创建集群元数据请求 MetadataRequest 对象
        MetadataRequest.Builder metadataRequest;
        if (metadata.needMetadataForAllTopics()) {
            // 需要更新所有 topic 的元数据信息
            metadataRequest = MetadataRequest.Builder.allTopics();
        } else {
            // 仅需更新指定 topic 的元数据信息
            metadataRequest = new MetadataRequest.Builder(new ArrayList<>(metadata.topics()));
        }

        // 将 MetadataRequest 包装成 ClientRequest 进行发送，在下次执行 poll 操作时一并发送
        log.debug("Sending metadata request {} to node {}", metadataRequest, node.id());
        sendInternalMetadataRequest(metadataRequest, nodeConnectionId, now);
        return requestTimeoutMs;
    }

    /* 不允许向目标节点发送请求 */

    // 如果存在到目标节点的连接，则等待一会，无需再次尝试创建新的连接
    if (isAnyNodeConnecting()) {
        return reconnectBackoffMs;
    }

    /* 如果不存在到目标节点连接 */

    // 如果允许创建到目标节点的连接，则初始化连接
    if (connectionStates.canConnect(nodeConnectionId, now)) {
        log.debug("Initialize connection to node {} for sending metadata request", node.id());
        initiateConnect(node, now); // 初始化连接
        return reconnectBackoffMs;
    }

    return Long.MAX_VALUE;
}

private void handleCompletedSends(List<ClientResponse> responses, long now) {
    for (Send send : this.selector.completedSends()) {
        // 获取缓存到 inFlightRequests 集合中的请求对象
        InFlightRequest request = this.inFlightRequests.lastSent(send.destination());
        // 检测请求是否期望响应
        if (!request.expectResponse) {
            // 当前请求不期望服务端响应，则从 inFlightRequests 集合中删除
            this.inFlightRequests.completeLastSent(send.destination());
            // 为当前请求生成 ClientResponse 对象
            responses.add(request.completed(null, now));
        }
    }
}

private void handleCompletedReceives(List<ClientResponse> responses, long now) {
    for (NetworkReceive receive : this.selector.completedReceives()) {
        // 获取返回响应的节点 ID
        String source = receive.source();
        // 从 inFlightRequests 集合中获取缓存的 ClientRequest 对象
        InFlightRequest req = inFlightRequests.completeNext(source);
        // 解析响应
        AbstractResponse body = parseResponse(receive.payload(), req.header);
        log.trace("Completed receive from node {}, for key {}, received {}", req.destination, req.header.apiKey(), body);
        if (req.isInternalRequest && body instanceof MetadataResponse) {
            // 如果是更新集群元数据对应的响应，则更新本地的缓存的集群元数据信息
            metadataUpdater.handleCompletedMetadataResponse(req.header, now, (MetadataResponse) body);
        } else if (req.isInternalRequest && body instanceof ApiVersionsResponse) {
            // 如果是更新 API 版本的响应，则更新本地缓存的目标节点支持的 API 版本信息
            this.handleApiVersionsResponse(responses, req, now, (ApiVersionsResponse) body);
        } else {
            // 否则，获取 ClientResponse 响应对象添加到队列中
            responses.add(req.completed(body, now));
        }
    }
}

RequestCompletionHandler callback = new RequestCompletionHandler() {
    @Override
    public void onComplete(ClientResponse response) {
        handleProduceResponse(response, recordsByPartition, time.milliseconds());
    }
};

private void completeBatch(RecordBatch batch, ProduceResponse.PartitionResponse response, long correlationId, long now) {
    Errors error = response.error;
    // 异常响应，但是允许重试
    if (error != Errors.NONE && this.canRetry(batch, error)) {
        log.warn("Got error produce response with correlation id {} on topic-partition {}, retrying ({} attempts left). Error: {}", correlationId, batch.topicPartition, retries - batch.attempts - 1, error);
        // 将消息重新添加到收集器中，等待再次发送
        this.accumulator.reenqueue(batch, now);
    }
    // 正常响应，或不允许重试的异常
    else {
        RuntimeException exception;
        if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
            // 权限认证失败
            exception = new TopicAuthorizationException(batch.topicPartition.topic());
        } else {
            // 其他异常，如果是正常响应，则为 null
            exception = error.exception();
        }
        // 将响应信息传递给用户，并释放 RecordBatch 占用的空间
        batch.done(response.baseOffset, response.logAppendTime, exception);
        this.accumulator.deallocate(batch);
    }

    // 如果是集群元数据异常，则标记需要更新集群元数据信息
    if (error.exception() instanceof InvalidMetadataException) {
        metadata.requestUpdate();
    }

    // 释放已经处理完成的 topic 分区，对于需要保证消息强顺序性，以允许接收下一条消息
    if (guaranteeMessageOrder) {
        this.accumulator.unmutePartition(batch.topicPartition);
    }
}

public void done(long baseOffset, long logAppendTime, RuntimeException exception) {
    log.trace("Produced messages to topic-partition {} with base offset offset {} and error: {}.", topicPartition, baseOffset, exception);

    // 标识当前 RecordBatch 已经处理完成
    if (completed.getAndSet(true)) {
        throw new IllegalStateException("Batch has already been completed");
    }

    // 设置当前 RecordBatch 发送之后的状态
    produceFuture.set(baseOffset, logAppendTime, exception);

    // 循环执行每个消息的 Callback 回调
    for (Thunk thunk : thunks) {
        try {
            // 消息处理正常
            if (exception == null) {
                // RecordMetadata 是服务端返回的
                RecordMetadata metadata = thunk.future.value();
                thunk.callback.onCompletion(metadata, null);
            }
            // 消息处理异常
            else {
                thunk.callback.onCompletion(null, exception);
            }
        } catch (Exception e) {
            log.error("Error executing user-provided callback on message for topic-partition '{}'", topicPartition, e);
        }
    }

    // 标记本次请求已经完成（正常响应、超时，以及关闭生产者）
    produceFuture.done();
}