吊打面试官|Kafka生产者架构深度剖析

// 这段代码展示了生产者的核心发送流程：
// 1. 消息发送入口 send() 方法
// 2. 消息拦截器的调用
// 3. 序列化和分区选择
// 4. 消息追加到累加器
// 5. 唤醒发送线程
//KafkaProducer.java 的核心发送方法
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    // 检查生产者是否已经关闭
    throwIfProducerClosed();
    
    // 开始记录发送时间
    long nowMs = time.milliseconds();
    
    // 调用拦截器
    ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
    
    // 执行序列化和分区选择
    return doSend(interceptedRecord, callback, nowMs);
}


// 异步发送的具体实现
private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback, long timestamp) {
    // 获取或创建topic元数据
    ClusterAndWaitTime clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs);
    Cluster cluster = clusterAndWaitTime.cluster;
    
    // 序列化key和value
    byte[] serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key());
    byte[] serializedValue = valueSerializer.serialize(record.topic(), record.headers(), record.value());
    
    // 选择分区
    int partition = partition(record, serializedKey, serializedValue, cluster);
    
    // 构建消息批次并追加到累加器
    RecordAccumulator.RecordAppendResult result = accumulator.append(...)
    
    // 如果需要，唤醒sender线程
    if (result.batchIsFull || result.newBatchCreated) {
        this.sender.wakeup();
    }
    
    return result.future;
}

// 这部分代码展示了：
// 1. 拦截器链的执行过程
// 2. 序列化器的接口定义
// 3. 具体序列化器的实现示例
// ProducerInterceptors.java
public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
    ProducerRecord<K, V> interceptRecord = record;
    for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
        try {
            interceptRecord = interceptor.onSend(interceptRecord);
        } catch (Exception e) {
            // 处理异常
        }
    }
    return interceptRecord;
}


// Serializer接口
public interface Serializer<T> extends Closeable {
    void configure(Map<String, ?> configs, boolean isKey);
    byte[] serialize(String topic, T data);
    byte[] serialize(String topic, Headers headers, T data);
    void close();
}


// StringSerializer示例
public class StringSerializer implements Serializer<String> {
    @Override
    public byte[] serialize(String topic, String data) {
        if (data == null)
            return null;
        return data.getBytes(StandardCharsets.UTF_8);
    }
}

// 分区器代码展示了：
// 1.默认分区策略的实现
// 2.基于key的hash分区
// 3.无key时的轮询分区
// DefaultPartitioner.java
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 {
            // 没有可用分区时，随机选择
            return Utils.toPositive(nextValue) % numPartitions;
        }
    } else {
        // 有key时，使用hash策略
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }
}

// 这段代码展示了：
// 消息批次的管理
// 批次大小的控制
// 消息追加的过程
// RecordAccumulator.java
public RecordAppendResult append(TopicPartition tp,
                               long timestamp,
                               byte[] key,
                               byte[] value,
                               Header[] headers,
                               Callback callback,
                               long maxTimeToBlock) {
    // 获取或创建消息批次
    ProducerBatch batch = getOrCreateBatch(tp, ...);
    
    // 追加记录到批次
    RecordAppendResult appendResult = batch.tryAppend(timestamp, key, value, headers, callback, time.milliseconds());
    
    // 处理追加结果
    if (appendResult != null) {
        return appendResult;
    }
    
    // 如果当前批次已满，创建新批次
    byte[] serializedKey = key;
    byte[] serializedValue = value;
    int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(magic, compression, serializedKey, serializedValue, headers));
    batch = createBatch(tp, size);
    
    // 再次尝试追加
    appendResult = batch.tryAppend(timestamp, key, value, headers, callback, time.milliseconds());
    
    return appendResult;
}

// 发送线程代码展示了：
// 发送循环的实现
// 批次的获取和发送
// 网络请求的处理
// Sender.java
public void run() {
    while (running) {
        try {
            runOnce();  // 执行一次发送循环
        } catch (Exception e) {
            log.error("Uncaught error in kafka producer I/O thread: ", e);
        }
    }
}


void runOnce() {
    // 获取准备发送的批次
    long currentTimeMs = time.milliseconds();
    long pollTimeout = sendProducerData(currentTimeMs);
    
    // 处理响应
    client.poll(pollTimeout, currentTimeMs);
}


private long sendProducerData(long now) {
    // 获取可发送的消息批次
    Map<Integer, List<ProducerBatch>> batches = accumulator.drain(...);
    
    // 发送消息批次
    for (Map.Entry<Integer, List<ProducerBatch>> entry : batches.entrySet()) {
        sendProducerData(entry.getKey(), entry.getValue());
    }
}

// 网络层代码展示了：
// 网络连接的管理
// 数据发送的实现
// 连接状态的维护
// NetworkClient.java
public boolean ready(Node node, long now) {
    // 检查连接状态
    if (node.isEmpty())
        return false;
    
    // 检查是否需要创建新连接
    if (connectionStates.canConnect(node.idString(), now))
        return true;
    
    // 检查连接是否就绪
    return connectionStates.isReady(node.idString(), now);
}


// KafkaChannel.java
public Send write() throws IOException {
    if (send == null)
        return null;
    
    int written = send.writeTo(transportLayer);
    if (send.completed())
        send = null;
    
    return send;
}

其他副本。集群的配置和部署直接影响系统的可用性、可靠性和扩展性。
// 这部分代码展示了：
// 生产者与集群的交互
// 元数据的获取和更新
// 集群状态的检查
// KafkaProducer与集群交互的相关代码
private ClusterAndWaitTime waitOnMetadata(String topic, Integer partition, long maxWaitMs) {
    // 获取集群元数据
    Cluster cluster = metadata.fetch();
    
    // 检查topic是否存在
    if (cluster.invalidTopics().contains(topic))
        throw new InvalidTopicException(topic);
    
    // 更新元数据
    metadata.add(topic);
    
    // 等待元数据更新
    cluster = metadata.awaitUpdate(version, maxWaitMs);
    
    return new ClusterAndWaitTime(cluster, remainingWaitMs);
}