凡人多烦事01

这个屌丝很懒，什么也没留下！

热门标签

Kafka Producer 实现源码分析

作者：凡人多烦事01 | 2024-05-24 17:16:29

踩

producerequestresult.await()阻塞

前言

拥抱变化接手了 Kafka 平台，遂学习 0.10.0 线上版本的设计与实现。限于篇幅，本文不会逐行解析源码，而是从逻辑流程、设计模式、并发安全等方面学习各组件，笔记仅供个人 Review

一：准备

1.1 配置项

参考文档 #producerconfigs，部分配置间会相互影响，如下：


batch.size      // 单个 batch 的最大字节数
linger.ms       // 控制 batch 未满时最多再等多久才发出
buffer.memory	// producer 消息缓冲区内存上限
max.block.ms	// 决定缓冲区满后阻发送一条消息的流程较长，涉及到的组件很多，塞等待可用内存的最长时间
retries	        // 发送消息出现临时性错误时的最大重试次数（默认无限重试以实现 AtLeastOnce）
max.in.flight.requests.per.connection	// 控制一条连接上能发送的在途请求数（默认 5），若不为 1 则在重试时可能导致消息乱序

乱序情况：


metadata.max.age.ms     // 集群元信息的有效时长，超过则强制刷新
connections.max.idle.ms // 连接的最长闲置时间，超过则主动断开
max.block.ms            // 应用层 send(), partitionFor() 的超时时间
request.timeout.ms      // 网络层任何请求等待响应的超时时间
retry.backoff.ms, reconnect.backoff.ms // 重发消息、重连 broker 的定时规避周期

实现

producer 有 20+ 配置项，配置模块需对用户给定的Map<String, Object>, Properties等键值对象，进行配置值的类型检查、有效性检查、默认值填充等处理，得到有效配置。配置模块：

ConfigDef：记录用户原始配置、解析后的配置，并提供 util 方法按类型读取配置值，特别地，define 系列方法都返回 this 以实现链式调用，类似于 Builder 模式
ProducerConfig.CONFIG单例：静态常量，并在 static 代码块中定义，故在类加载阶段就会被初始化，是典型的单例模式应用，类似单例还有枚举实现的 protocol.ApiKeys,protocol.Errors 等等
Configurable接口规范类的反射构造行为：用户类（Partitioner, Serializer, Interceptor…）配置值都是字符串，反射实例化时构造方法可能无参或有参但类型不定，故抽象出Configurable接口，无参构造完成后接收配置值，进一步实例化


public <T> T getConfiguredInstance(String key, Class<T> t) {
  Class<?> c = getClass(key);
  Object o = Utils.newInstance(c); // 无参构造实例化
  if (o instanceof Configurable)
      ((Configurable) o).configure(this.originals); // 完善构造
  return t.cast(o);
}

1.2 预处理

如下 demo 展示了两种发送消息的方式

send 异步发送：返回类型为 RecordMetadata 的 Future 对象，调用 get 等待消息的发送结果
callback 异步发送：实现 Callback 接口，异步处理发送结果


public class Main {
    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
 
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        ProducerRecord<String, String> record = new ProducerRecord<>("topic-01", "msg1");
        Future<RecordMetadata> future = producer.send(record);
        RecordMetadata meta = future.get(1, TimeUnit.SECONDS);
        log.info("[invoke] send succeed, offset: {}", meta.offset());
 
        producer.send(record, new UserCallback());
        producer.flush();
    }
 
    static class UserCallback implements org.apache.kafka.clients.producer.Callback {
        @Override
        public void onCompletion(RecordMetadata metadata, Exception exception) {
            if (exception == null) // 无异常，发送成功
                log.info("[callback] send succeed, offset: {}", metadata.offset());
            else // 发送失败
                log.info("[callback] send failed, exception: {}", exception.getMessage());
        }
    }
}
 
// INFO [invoke] send succeed, offset: 12 (org.apache.kafka.clients.producer.Main:24)
// INFO [callback] send succeed, offset: 13 (org.apache.kafka.clients.producer.Main:34)

底层发送流程如下，之后章节将逐一解析各个组件：

1. ProducerInterceptors

用户类可线程安全地实现ProducerInterceptor接口（限制不能抛异常），用在

消息发送前修改内容：如统一添加 msg uuid
在返回前读取元数据：如记录异常信息，tracing 日志等

支持组合多个拦截器，批量链式调用，但不提倡，因为链式调用时若某个拦截器抛出 unchecked 异常，捕捉后仅记录日志不抛回 send 调用方，用户无法感知


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) {
            // onSend 抛出运行时异常，只会打 warn 日志，本次拦截视为无效
            // 如 A->B->C->D，若 C 拦截时抛异常，则 D 拿到的消息是 B 拦截后的，这种行为非预期
            log.warn("Error executing interceptor onSend callback", e); // ...
        }
    }
    return interceptRecord;
}

2. Serializer

发送消息的 Key 和 Value 统一用byte[]描述，实现二进制安全，故用户类需实现Serializer接口；serialization 模块已内置了基础数字类型、String 的序列化实现，consumer 反序列化 Deserializer 接口同理


public interface Serializer<T> extends Closeable {
    public byte[] serialize(String topic, T data); /*...*/
}

3. Partitioner

用户实现Partitioner来决定每一条消息要发往哪个分区


public interface Partitioner extends Configurable {
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);
}

默认分区器DefaultPartitioner选择分区的流程：

至此，分析了发送前消息的拦截修改、键值序列化、确定分区的逻辑

二：内存层

职责：内存池资源管理，消息压缩与 batch 分批，发送结果的异步计算

2.1 BufferPool

buffer.memory 配置限制了 producer 缓冲消息所能使用的最大内存，默认启用 batch 机制后，消息常以batch.size大小分批发送，故设计内存池重用 batch 内存

内存池划分为 3 个区域：

free 双端队列：元素为batch.size大小的内存块，消息发送成功后 clear 入队重用；有 2 个特性
- 惰性分配：队列初始状态为空，后续分配的 batch “不规整内存块” 使用完毕则入队
- 动态回收：若 availableMemory 内存不足，但 free 队列有空闲内存块时，会逐个出队释放内存（GC）
不规整内存块：若 batch 被禁用，或发送大小在(batch.size, max.request.size]范围的大消息，会直接分配该大小的一次性 ByteBuffer，使用完毕后由 GC 回收
空闲内存：非内存块实体，只是统计值，在分配和释放不规整内存块时对应增减库存

内存不足时，调用send()的多个用户线程都会 await 阻塞在各自的条件变量上，内存池采用先到先得的策略，当有内存可用时只会 signal 唤醒入队最早、等待时间最长的线程，避免了线程饥饿或多线程低效竞争。示意图：

如上 thread_0 被唤醒后，会收集该可用内存，若内存已足够则恢复运行并唤醒 thread_1，否则继续等待

2.2 RecordBatch

消息批次用 RecordBatch 描述，维护消息重发、future 结果等元信息，消息实际存储在底层的 MemoryRecords 缓冲区，并使用 Compressor 进行压缩

1. Compressor

1）类加载

producer 支持三种压缩方式：gzip, snappy 和 lz4，但只有 gzip 由 java.util.zip JDK 标准库实现，其他 2 种压缩类需添加 jar 包，在运行时反射加载，不使用时能减少包体积；同时为保证 producer 全局只会反射构造出一个 Constructor，用到了懒加载的 DCL 单例模式

2）封装 batch

每种算法都有预期压缩比，如 gzip 是 50%；在 MemoryRecords 视角，将 16KB batch 传给 Compressor 后，实际至多写入 32KB 数据，由于压缩比不精确，Compressor 要有动态扩容的能力，以容纳更多压缩消息；实现：


public class ByteBufferOutputStream extends OutputStream {
    /*...*/
    public void write(int b) {
        if (buffer.remaining() < 1)
            expandBuffer(buffer.capacity() + 1); // 内存不足，自动扩容
        buffer.put((byte) b);
    }
    private void expandBuffer(int size) { // 容量增长 10%，或增长到 size 字节
        int expandSize = Math.max((int) (buffer.capacity() * REALLOCATION_FACTOR), size);
        ByteBuffer temp = ByteBuffer.allocate(expandSize);
        temp.put(buffer.array(), buffer.arrayOffset(), buffer.position());
        buffer = temp;
    }
}

3）压缩实现

实现了两层装饰模式：为 ByteBuffer 装饰了自动扩容功能，为各种类型数据的 put 操作装饰了压缩写：

特别地，putRecord() 带压缩地写入一条消息，写入后的内存结构：

更正：magic number 为 1 时带 timestamp，为 0 则无 timestamp 字段

缓冲区写满后触发close()，Compressor 会倒回至 ByteBuffer 的初始位置，插入一条 Shadow Record 补全元数据，将其后所有压缩后的 Real Records 视为其 key 的值，内存结构：

TODO: 勘误画图，压缩后的 RealRecords 放的是 value 字段

注意，close() 方法中会动态调整压缩比，即压缩比是自适应的

4）写满判断

判断 MemoryRecords 是否已满，是通过估算 Compressor 压缩后字节数实现的，估算逻辑：


public class Compressor {
    public long estimatedBytesWritten() {
        if (type == CompressionType.NONE)
            return bufferStream.buffer().position();
        else // 有压缩，压缩后字节数 = 未压缩字节数 * 压缩比 * 误差因子
             // 比如用 gzip 压缩 1MB 数据：1MB*0.5*1.05 = 537KB
            return (long) (writtenUncompressed * TYPE_TO_RATE[type.id] * COMPRESSION_RATE_ESTIMATION_FACTOR);
    }
}
 
public class MemoryRecords implements Records {
    public boolean isFull() { // 估算压缩后的大小是否已超 Compressor 容量
        return !this.writable || this.writeLimit <= this.compressor.estimatedBytesWritten();
    }
}

2. MemoryRecords

负责委托 Compressor 追加写 Record，为其添加 LOG_OVERHEAD 头信息


public long append(long offset, long timestamp, byte[] key, byte[] value) {
    if (!writable)
        throw new IllegalStateException("Memory records is not writable");
    int size = Record.recordSize(key, value);
    compressor.putLong(offset); // 此 record 在 batch 中的相对偏移量
    compressor.putInt(size);    // 压缩前此 record 的大小
    long crc = compressor.putRecord(timestamp, key, value);
    compressor.recordWritten(size + Records.LOG_OVERHEAD);
    return crc;
}

当缓冲区满后会切换为只读模式，等待 drain 选中发出


public void close() {
    if (writable) {
        compressor.close();                // 回填压缩元数据
        this.buffer = compressor.buffer(); // compressor 持有的 ByteBuffer，可能已扩容，更新指向
        this.buffer.flip();                // 切为读模式
        writable = false;
    }
}

基于以上 2 个组件，RecordBatch 实现了三个机制：

1）委托 MemoryRecords 追加写 Record，并将各 Record 的元数据 future 与 batch 写结果相关联


public FutureRecordMetadata tryAppend(long timestamp, byte[] key, byte[] value, Callback callback, long now) {
    /*...*/
    // 递增 offsetCounter，即递增 record 的 relative offset，追加写到 batch 中
    long checksum = this.records.append(offsetCounter++, timestamp, key, value);
    FutureRecordMetadata future = new FutureRecordMetadata(this.produceFuture, this.recordCount,
                                                           timestamp, checksum,
                                                           key == null ? -1 : key.length,
                                                           value == null ? -1 : value.length);
    if (callback != null)
        thunks.add(new Thunk(callback, future)); // 用户 callback 组合 record future 记为 thunk
    return future;
}

2）batch 发送结束后，发送结果 baseOffset 和 exception 会被填充到 ProduceRequestResult，通知各 Record 的 FutureRecordMetadata，唤醒阻塞在 get() 调用上的用户线程：

此处多线程需等待单线程执行结果，用 CountDownLatch 模拟实现了 Future 接口：


public final class ProduceRequestResult {
    private final CountDownLatch latch = new CountDownLatch(1); // 模拟 future 异步通知与等待
    public void done(TopicPartition topicPartition, long baseOffset, RuntimeException error) {
        this.topicPartition = topicPartition;
        this.baseOffset = baseOffset;
        this.error = error;
        this.latch.countDown(); // 发送完毕
    }
    public void await() throws InterruptedException {
        latch.await();
    }
}
 
public final class FutureRecordMetadata implements Future<RecordMetadata> {
    private final ProduceRequestResult result; // 二者是组合关系
    private final long relativeOffset;
 
    @Override
    public RecordMetadata get() throws InterruptedException, ExecutionException {
        this.result.await(); // 阻塞等待 batch 发送完毕，会在 Sender.handleProduceResponse 中调用 done
        // baseOffset + relativeOffset 即此消息在分区中的绝对 offset
        return new RecordMetadata(result.topicPartition(), this.result.baseOffset(), this.relativeOffset, /*...*/);
    }
}

3）维护 batch 的重发状态

attempts：已重试次数，在retries内都会重试
lastAttemptMs：上次重试发送的时间戳，配合retry.backoff.ms避免频繁重试

2.3 RecordAccumulator

accumulator 维护各 topic partition（tp）的 batch 队列，结构如下：


public final class RecordAccumulator {
    private final BufferPool free;                                           // 内存池
    private final ConcurrentMap<TopicPartition, Deque<RecordBatch>> batches; // 每个 topic 分区持有一个 RecordBatch 双端队列
}

producer 的 send 操作只将消息放入对应的 RecordBatch 中即返回：

accumulator 还负责对各 tp 的 batch 队列进行 rollover，在实现上有两个优化亮点：

（1）细粒度锁，提高可用性：创建新 batch 时需阻塞申请内存，会主动放弃 dq 互斥锁。示意图：

问题：A 持有 dq 锁，阻塞申请内存，会导致虽然内存够，但 B 也必须等待
解决：当锁范围内有耗时操作时，考虑拆为细粒度锁，减少锁的持有时间

（2）解决并发 rollover 的内存碎片问题：细粒度锁的副作用是引入了新的并发竞争

问题：A,B 同时发送大消息，都创建新 batch 后都入队，先入队的 batch 不再被使用，剩余内存将被浪费
解决：创建完新 batch 后不着急使用，先尝试写入 last batch，若写入成功则释放新 batch

缓存消息的实现：


public RecordAppendResult append(TopicPartition tp, long timestamp, byte[] key, byte[] value, Callback callback, long maxTimeToBlock) {
    // 1. 获取 tp 的可用 batch
    Deque<RecordBatch> dq = getOrCreateDeque(tp); // 获取或创建此 tp 的 batch 队列
    synchronized (dq) {                           // batch deque 并非线程安全，整个队列加锁
        RecordBatch last = dq.peekLast();         // 之前 batch 都已满，取最后一个 batch
        if (last != null) {
            FutureRecordMetadata future = last.tryAppend(timestamp, key, value, callback, time.milliseconds());
            if (future != null) { // 写成功直接返回
                return new RecordAppendResult(future, dq.size() > 1 || last.records.isFull(), false);
            }    
        }
    } // 放弃 dq 锁
 
    // 2. tp 的 batch 队列为空，或现有 batch 都已满，阻塞申请内存创建新 batch
    int size = Math.max(this.batchSize, Records.LOG_OVERHEAD + Record.recordSize(key, value));
    ByteBuffer buffer = free.allocate(size, maxTimeToBlock);
 
    synchronized (dq) {                   // 重新加锁 
        RecordBatch last = dq.peekLast(); // 重新尝试入队 last
        if (last != null) {
            FutureRecordMetadata future = last.tryAppend(timestamp, key, value, callback, time.milliseconds());
            if (future != null) {
                free.deallocate(buffer);  // 使用 batchA，释放 batchB
                return new RecordAppendResult(future, dq.size() > 1 || last.records.isFull(), false);
            }
        }
        // 3. 构建新 batch 重新入队，肯定能写成功
        MemoryRecords records = MemoryRecords.emptyRecords(buffer, compression, this.batchSize);
        RecordBatch batch = new RecordBatch(tp, records, time.milliseconds());
        FutureRecordMetadata future = Utils.notNull(batch.tryAppend(timestamp, key, value, callback, time.milliseconds()));
        dq.addLast(batch); // 新 batch 入队
 
        return new RecordAppendResult(future, dq.size() > 1 || batch.records.isFull(), true);
    }
}

至此，分析了消息的内存分配、压缩写入、batch 读写模式切换、batch rollover 等机制

三：网络层

职责：维护连接状态，执行四种网络 IO 并收集结果

producer 网络层使用了 NIO Selector 机制，内部各组件关系如下，逐个分析

3.1 KafkaChannel

1. TransportLayer

封装 SocketChannel 的读写，提供注册事件的快捷方法


// 实现分割、聚合读写 ByteBuffer 的 Channel
interface TransportLayer extends ScatteringByteChannel, GatheringByteChannel{ /*...*/ }
 
public class PlaintextTransportLayer implements TransportLayer { // 明文传输层
    private final SelectionKey key;
    private final SocketChannel socketChannel;
 
    @Override
    public boolean finishConnect() throws IOException {
        boolean connected = socketChannel.finishConnect(); // 等待连接建立，并自动注册 OP_READ 事件
        if (connected)
            key.interestOps(key.interestOps() & ~SelectionKey.OP_CONNECT | SelectionKey.OP_READ);
        return connected;
    }
 
    @Override
    public int read(ByteBuffer dst) throws IOException {
        return socketChannel.read(dst); // 读写操作都被直接委托给了底层的 SocketChannel
    }
    public int write(ByteBuffer src) throws IOException {
        return socketChannel.write(src);
    }
    
    @Override
    public void addInterestOps(int ops) {
        key.interestOps(key.interestOps() | ops); // 注册新事件
    }
    /*...*/
} /* TODO: SslTransportLayer */

2. 读写缓冲区

读缓冲 NetworkReceive 类：从 SocketChannel 中拆包，持续读取一个完整的响应
写缓冲 Send 接口：由 RequestSend->NetworkSend->ByteBufferSend 装饰链，将一个完整的请求封包，持续写入 SocketChannel

KafkaChannel 只维护一个读缓冲、一个写缓冲，并提供对应的read, write方法读写 TransportLayer

3.2 KafkaSelector

在 NIO Selector 上封装了网络 IO 的单次执行 4 种结果，对应了 4 种 IO 事件：


List<String> connected;                 // OP_CONNECT    // 新建连接的 broker id 列表
List<String> disconnected;              // closed, error // 断开连接的 broker id 列表
List<NetworkReceive> completedReceives; // OP_READ       // 读取到的响应数据
List<Send> completedSends;              // OP_WRITE      // 已写出的请求数据

执行网络 IO：


public void poll(long timeout) throws IOException {
    clear(); // 清理上次 poll 的结果
    int readyKeys = select(timeout); // 等待网络 IO
    if (readyKeys > 0)
        pollSelectionKeys(this.nioSelector.selectedKeys(), false);
    addToCompletedReceives();
}
 
private void pollSelectionKeys(Iterable<SelectionKey> selectionKeys, boolean isImmediatelyConnected) {
    Iterator<SelectionKey> iterator = selectionKeys.iterator();
    while (iterator.hasNext()) { 
        SelectionKey key = iterator.next();  // 逐个处理 selectionKey
        iterator.remove();
        KafkaChannel channel = channel(key); // 取出 attach 的 kafkaChannel
 
        try {
            // 1. 处理 OP_CONNECT
            if (isImmediatelyConnected || key.isConnectable()) {
                if (channel.finishConnect())          // 连接成功，注册 OP_READ
                    this.connected.add(channel.id()); // 1. 收集新连接
                else
                    continue; // 连接未完成，后续不必再处理
            }
 
            // 2. 处理 OP_READ
            if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
                NetworkReceive networkReceive;
                while ((networkReceive = channel.read()) != null)
                    addToStagedReceives(channel, networkReceive); // 2. 收集一个完整的 Receive 响应
            }
 
            // 3. 处理 OP_WRITE
            if (channel.ready() && key.isWritable()) {
                Send send = channel.write(); // 写出 send 缓冲
                if (send != null)
                    this.completedSends.add(send); // 3. 收集写成功的 Send 请求
            }
 
            // 4. 处理 closed
            if (!key.isValid()) {
                close(channel);
                this.disconnected.add(channel.id()); // 4. 收集不可用连接
            }
        } catch (Exception e) {
            close(channel);
            this.disconnected.add(channel.id());  // 有异常视为连接不可用
        }
    }
}

结果收集完毕后，通过 List<Send> completedSends()等多个对应方法暴露给上层的 NetworkClient

3.3 NetworkClient

负责统一网络 IO 结果并解析响应，维护发往各个节点的有序请求队列，维护与各节点单一连接的状态；持有 2 个子组件：

（1）ClusterConnectionStates：连接状态管理

client 与每个 broker 都只会保持一条 TCP 连接，而非维护一个连接池，以简化消息有序性的实现。连接有三种状态，并由NodeConnectionState维护重连信息，由ClusterConnectionStates持有整个集群的连接状态


final class ClusterConnectionStates {
    private final long reconnectBackoffMs;
    private final Map<String, NodeConnectionState> nodeState;
    /* ... */  
    private static class NodeConnectionState {
        ConnectionState state;
        long lastConnectAttemptMs;
    }
}
 
public enum ConnectionState {
    DISCONNECTED, CONNECTING, CONNECTED
}

（2）InFlightRequests：各节点的有序请求队列

client 发往各节点的请求都会对应入队 InFlightRequests 暂存，以实现三个功能：

有序收发：收到响应的顺序，必须与发出请求的顺序保持一致
请求超时检测：发出的请求在request.timeout.ms时限仍未收到响应，向用户返回连接异常
请求数限制：限制单个节点（连接）并发请求数，不超过max.in.flight.requests.per.connection配置，配置设为 1 以实现消息的绝对有序性


final class InFlightRequests {
    private final int maxInFlightRequestsPerConnection;
    private final Map<String, Deque<ClientRequest>> requests = new HashMap<String, Deque<ClientRequest>>();
 
    // 是否限制 node 发送新请求
    public boolean canSendMore(String node) {
        Deque<ClientRequest> queue = requests.get(node);
        return queue == null || queue.isEmpty() || // 1. 未发出任何请求
                    // 2. 上一个请求必须已完成 write，否则网络可能不可用，上一个请求还在该 channel 的 send 缓冲区中，不能强行覆盖
                    (queue.peekFirst().request().completed() 
                    // 3. 同时在途请求数不能超过配置
                    && queue.size() < this.maxInFlightRequestsPerConnection);
               
    }
    // 筛出出有超时请求的节点列表
    public List<String> getNodesWithTimedOutRequests(long now, int requestTimeout) {
        List<String> nodeIds = new LinkedList<String>();
        for (String nodeId : requests.keySet()) {
            if (inFlightRequestCount(nodeId) > 0) {
                ClientRequest request = requests.get(nodeId).peekLast(); // 最早入队的 request
                long timeSinceSend = now - request.sendTimeMs();
                if (timeSinceSend > requestTimeout) // 超过 request.timeout.ms 都还未处理完，视为超时
                    nodeIds.add(nodeId);
            }
        }
        return nodeIds;
    }
}

此外，还负责读取网络 IO 结果，执行协议解析并汇总成 Response pipeline，逐个调用 protocol handler：


public List<ClientResponse> poll(long timeout, long now) {
    long metadataTimeout = metadataUpdater.maybeUpdate(now); // 若有必要，发起额外的 MetadataRequest 请求
    
    this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs)); // min timeout 保证各种超时事件能被及时执行
    List<ClientResponse> responses = new ArrayList<>();
 
    // 1. 处理 acks=0 的请求：对于发送成功的 send，若 acks=0 不需要等待响应，构造空 response 并收集
    handleCompletedSends(responses, updatedNow);
 
    // 2. 解析响应：用对应版本的协议，去解析 receive 响应数据，填充到 Struct 后入队 responses
    handleCompletedReceives(responses, updatedNow);
 
    // 3. 处理连接断开：已断开连接节点的 in-flights 请求，构造 disconnect response
    handleDisconnections(responses, updatedNow);
 
    // 4. 处理新建连接
    handleConnections();
 
    // 5. 处理请求超时：超时节点被视为不可用，同样为 in-flights 队列重的请求，构造 disconnect response
    // 连接断开是网络异常，请求超时是网络延迟，都视为节点不可用，连接都会被置为 DISCONNECTED 等待重连
    handleTimedOutRequests(responses, updatedNow);
 
    // 处理本轮 IO 结果，调用 request 的 callback handler
    for (ClientResponse response : responses) {
        if (response.request().hasCallback())
            // 如 Produce 请求会执行 RequestCompletionHandler.handleProduceResponse
            response.request().callback().onComplete(response);
    }
    return responses;
}

至此，分析了底层 SocketChannel 的读缓冲拆包、写缓冲封包，中间层 NIO Selector 四种网络 IO 事件的处理及单次结果收集，上层 NetworkClient 维护连接状态及 IO 结果处理等机制

四：数据处理层

职责：筛选出待发送的 batch，构造请求，处理发送结果，重发消息，维护集群元数据

4.1 通信协议

（1）协议描述

参考文档#The_Messages_Produce，以 Produce v0 请求为例，协议的字段分布如下：


acks => INT16
timeout_ms => INT32
[  topic_name => STRING # [] 表示数组
   topic_data => [
      partition_index => INT32
      record_set      => BYTES # 压缩后的 batch 队列数据
   ]
]

可见协议是由类型不一的字段组合嵌套而成，protocol 模块用Type类描述字段类型，Field类描述字段本身，Schema类描述协议字段集，Struct描述协议及对应数据：


public class Field {
    public final String name; // 字段名
    public final Type type;   // 可序列化的字段类型
}
 
public abstract class Type {
    public abstract void write(ByteBuffer buffer, Object o); // 每种数据类型都需实现的读写方法
    public abstract Object read(ByteBuffer buffer);          // 即完成数据的序列化、反序列化
    public static final Type INT32 = new Type() {/*...*/}    // 基本类型、String、ByteBuffer 对应的协议类型
    public static final Type BYTES = new Type() {/*...*/}
}
 
public class Schema extends Type {
    private final Field[] fields; // 协议的字段集
    private final Map<String, Field> fieldsByName;
}
public class Struct {
    private final Schema schema;
    private final Object[] values; // 协议对应的数据
}

由上可知 Produce v0 请求可被描述为：


public static final Schema TOPIC_PRODUCE_DATA_V0
= new Schema(new Field("topic", STRING),
             new Field("data", new ArrayOf(new Schema(new Field("partition", INT32),
                                                      new Field("record_set", BYTES)))));
 
public static final Schema TOPIC_PRODUCE_DATA_V0
= new Schema(new Field("acks", INT16),
             new Field("timeout", INT32),
             new Field("topic_data", new ArrayOf(TOPIC_PRODUCE_DATA_V0)));

（2）统一协议头

client 发出的每个请求都会带上 header，标识此请求的版本、类型、自增 id，描述如下：


public static final Schema REQUEST_HEADER 
= new Schema(new Field("api_key", INT16),        // 请求类型，比如 Metadata 类型为 3
             new Field("api_version", INT16),    // 协议版本号，如 v2 版本值为 2
             new Field("correlation_id", INT32), // 请求的自增唯一 id，broker 会原样返回，用于校验响应
             new Field("client_id", NULLABLE_STRING));
 
public enum ApiKeys {
    PRODUCE(0, "Produce"),
    METADATA(3, "Metadata"), /* ... */
}

4.2 Metadata

（1）元信息

producer 需知道各 topic 的分区分布、各分区的 leader broker 地址，此类元信息由 Cluster 类描述：


public class Node { // broker 由 Node 描述
    private final int id;
    private final String idString;
    private final String host;
    private final int port;
}
public class PartitionInfo {
    private final String topic; // TopicPartition: Topic, Partition
    private final int partition;
    private final Node leader;
}
 
public final class Cluster {
    private final List<Node> nodes;
    private final Map<Integer, Node> nodesById;                       // nodeId -> node
    private final Map<Integer, List<PartitionInfo>> partitionsByNode; // leaderNode -> [partInfo, ...]
    private final Map<TopicPartition, PartitionInfo> partitionsByTopicPartition; // tp -> partInfo
    private final Map<String, List<PartitionInfo>> partitionsByTopic;            // topic -> [all tp]
    private final Map<String, List<PartitionInfo>> availablePartitionsByTopic;   // topic -> [有 leader 的 tp]
}

类比 String，Cluster 类及其字段都是 final 只读，被修改时返回新实例，以保证 Sender 线程写、Producer 主线程读不存在并发问题。加上版本号分辨元信息的新旧，同时维护重试状态，由 Metadata 类描述：


public final class Metadata {
    private int version; // 版本号标识 Cluster 的新旧
    private Cluster cluster;
    private long lastRefreshMs;           // 上次更新时间戳，用于重试 backoff 等待
    private long lastSuccessfulRefreshMs; // 上次成功更新的时间戳
    private boolean needUpdate;       // 是否需要强制更新
}
/* 各种 synchronized 读写方法*/

（2）更新时机

NetworkClient 内部有一个持有 Metadata 的 DefaultMetadataUpdater 类，负责发起 MetadataRequest 更新请求并解析响应。client 每次执行 poll 前，都会先检查是否需要更新 Metadata，由于更新操作会阻塞主线程，故触发条件较为苛刻，有两层筛选：

Metadata 层：需等待自动过期、等待 backoff、被请求强制更新（网络层新建连接、断开连接、请求超时）
DefaultMetadataUpdater 层：若网络层无节点可用，不具备更新条件，也要等待 backoff

（3）更新实现

Metadata 由 Sender 线程写，而多个 Producer 线程读，是典型的线程间通信场景，故使用同一个 Sender.metadata对象的wait & notifyAll实现

实现：


public class KafkaProducer<K, V> implements Producer<K, V> {
    private long waitOnMetadata(String topic, long maxWaitMs) throws InterruptedException {
        /* ... */
        while (metadata.fetch().partitionsForTopic(topic) == null) { // 带超时地等待该 topic 元数据可用
            int version = metadata.requestUpdate();         // 请求更新
            sender.wakeup();                                // 唤醒 selector，触发 NetworkClient 执行 poll
            metadata.awaitUpdate(version, remainingWaitMs); // 等待 Sender 线程更新完毕调用 notifyAll
        }
    }
}
 
public final class Metadata {    
    public synchronized void awaitUpdate(final int lastVersion, final long maxWaitMs) throws InterruptedException {
        while (this.version <= lastVersion) {
            if (remainingWaitMs != 0)
                wait(remainingWaitMs);
            /*...*/
        }
    }
    
    public synchronized void update(Cluster cluster, long now) {
        this.version += 1;
        this.cluster =  cluster;
        notifyAll();
    }
}
 
public class NetworkClient implements KafkaClient {
    class DefaultMetadataUpdater implements MetadataUpdater {
        private final Metadata metadata;
        /*...*/ // 找出 in-flights 请求数最少的节点，发起 Metadata 请求
        private void handleResponse(RequestHeader header, Struct body, long now) {
            MetadataResponse response = new MetadataResponse(body);
            Cluster cluster = response.cluster(); // 解析响应读取元信息
            this.metadata.update(cluster, now);   // 通知 Producer 线程
        }
    }
}

4.3 Sender

（1）筛选各 node 需要发出的 batch 队列

Sender 线程在读写网络层之前，会根据 accumlator 排队时间、网络状态等条件，筛选出最紧急、最可能发送成功的 batch 集合，核心实现：


public class Sender implements Runnable {
    void run(long now) {
        // 1. 读取缓存的集群元信息
        Cluster cluster = metadata.fetch();
        // 2. accumulator: 筛选出有 topic 分区 batch 要发的节点集合
        RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);
        // 3. kafkaClient: 筛掉出符合网络 IO 条件的节点集合
        Iterator<Node> iter = result.readyNodes.iterator();
        while (iter.hasNext()) 
            /*...*/ // 过滤非 CONNECTED 连接
        // 4. accumulator：受限于请求大小上限，为各 node 收集其负责读写的各 tp 上的等待时间最长 batch
        Map<Integer, List<RecordBatch>> batches = this.accumulator.drain(cluster, result.readyNodes, this.maxRequestSize, now);
        // 5. 检测已超时的 batch
        this.accumulator.abortExpiredBatches(this.requestTimeout, now);
        // 6. 将 batch 队列对于封装为 ClientRequest 队列
        List<ClientRequest> requests = createProduceRequests(batches, now);
        for (ClientRequest request : requests) 
            client.send(request, now); // 发送
        // 7. 发送请求，读取响应
        this.client.poll(pollTimeout, now); // pollTimeout 计算取了节点重连，lingerMs 到期等时间最小值
    }
}

如上有两个核心筛选条件

ready()：根据各 tp 的 batch 队列缓存情况，筛选出有 batch 要发送的 node 集合
drain()：每个 node 需发送多个 tp batch 队列中的最老 batch，但由于单个请求大小有max.request.size上限，为避免分区饥饿（有的分区迟迟不被选中导致 batch 超时），会从随机的 tp 开始收集 batch；若 max.request.size 较大，还会继续收集各个 tp 等待时间第二长的 batch，设计十分巧妙！

（2）协议封装

第一步为各 node 都筛选出了要发送的 batch 队列，还需进一步封装为 ClientRequest：


// 为各 node 构建 ClientRequest 网络请求
private ClientRequest produceRequest(long now, int destination, short acks, int timeout, List<RecordBatch> batches) {
    Map<TopicPartition, ByteBuffer> produceRecordsByPartition = new HashMap<TopicPartition, ByteBuffer>(batches.size());
    final Map<TopicPartition, RecordBatch> recordsByPartition = new HashMap<TopicPartition, RecordBatch>(batches.size());
    for (RecordBatch batch : batches) {
        TopicPartition tp = batch.topicPartition;
        produceRecordsByPartition.put(tp, batch.records.buffer()); // tp -> 压缩后的 byteBuffer
        recordsByPartition.put(tp, batch);                         // tp -> 维护重试信息的 RecordBatch
    }
    // ProduceRequest 协议及数据 --> RequestSend 写缓冲
    ProduceRequest request = new ProduceRequest(acks, timeout, produceRecordsByPartition);
    RequestSend send = new RequestSend(Integer.toString(destination),
                                       this.client.nextRequestHeader(ApiKeys.PRODUCE), // 单调递增 correlation id
                                       request.toStruct());
 
    // 当 NetworkClient 收到此请求的响应时，将调用此 callback 处理响应结果
    RequestCompletionHandler callback = new RequestCompletionHandler() {
        public void onComplete(ClientResponse response) {
            handleProduceResponse(response, recordsByPartition, time.milliseconds()); // 捕捉了各 tp 的 batch 信息
        }
    };
 
    // RequestSend --callback handler--> ClientRequest
    return new ClientRequest(now, acks != 0, send, callback);
}

（3）处理发送结果

在 NetworkClient 收到响应后，会执行先解析出响应的 Struct，协议如下：


public static final Schema PRODUCE_RESPONSE_V2
= new Schema(new Field("responses",
             new ArrayOf(new Schema(new Field("topic", STRING),
                                    new Field("partition_responses",
                                    new ArrayOf(new Schema(new Field("partition", INT32),
                                                           new Field("error_code", INT16),
                                                           new Field("base_offset", INT64),
                                                           new Field("timestamp", INT64))))))),
              new Field("throttle_time_ms", INT32));

返回结果中指明了各 tp 的 base_offset 与 error_code，若有错误则检查是否可重试，无错误则 batch 发送成功


private void completeBatch(RecordBatch batch, Errors error, long baseOffset, long timestamp, long correlationId, long now) {
    if (error != Errors.NONE && canRetry(batch, error)) {
        this.accumulator.reenqueue(batch, now); // 1. 发生可重试错误，重新入队到队头，尽快重发
    } else {
        // 2. 致命错误 or 重试次数已耗尽 or 无错误
        RuntimeException exception;
        if (error == Errors.TOPIC_AUTHORIZATION_FAILED)
            exception = new TopicAuthorizationException(batch.topicPartition.topic());
        else
            exception = error.exception();            // 各种异常单例
        batch.done(baseOffset, timestamp, exception); // 回调用户 callback，构造 future 结果结束等待
        this.accumulator.deallocate(batch);           // 向 accumulator 归还 batch 内存块
    }
    if (error.exception() instanceof InvalidMetadataException)
        metadata.requestUpdate(); // 元数据过期，请求强制更新
    if (guaranteeMessageOrder)
        this.accumulator.unmutePartition(batch.topicPartition); // 解除 tp 屏蔽
}

至此，分析了协议描述，Metadata 更新机制，accumulator 筛选 batch 的两层过滤机制，以及 Sender 包装请求和处理发送结果的过程

总结

本文将 Kafka Producer 分为了三层

内存层：Compressor 实现消息压缩；MemoryRecords 实现 batch 写入并分批；RecordBatch 实现 batch 中各条消息元数据的异步计算，维护消息重发元数据；RecordAccumulator 则负责内存分配与协调
网络层：KafkaChannel 在 SocketChannel 上封装了拆包的读缓冲、封包的写缓冲；KafkaSelector 负责执行网络 IO 并收集结果；NetworkClient 负责维护连接状态，解析 IO 结果
数据处理层：Sender 线程从内存层筛选 batch，构造 Produce 请求下发给网络层，并处理发送结果

Producer 的亮点很多，个人认为有三点

朴素的并发设计：用 Condition 队列实现公平的内存分配、用 CountdownLatch 简化实现 Future 异步通知机制、用 metadata 对象的 wait & notifyAll 实现多线程同步等待更新，用 DCL 思想反射实例化 Compressor Constructor 单例…
严谨的并发逻辑：RecordAccumulator 在 rollover batch 时解决了细粒度锁引入的内存碎片问题…
简洁的模块解耦：RecordAccumulator 负责消息的缓冲分批，Sender 负责筛选 batch 构造发送请求并处理发送结果，NetworkClient 负责执行网络 IO

本文分析了 send 流程涉及到的核心模块及部分代码，更细致的逻辑还需参考源码

原文地址：https://yinzige.com/2020/02/15/kafka-producer/

end


Flink 从入门到精通 系列文章
 
基于 Apache Flink 的实时监控告警系统
关于数据中台的深度思考与总结（干干货）
日志收集Agent，阴暗潮湿的地底世界

公众号(zhisheng)里回复 面经、ClickHouse、ES、Flink、 Spring、Java、Kafka、监控 等关键字可以查看更多关键字对应的文章。

点个赞+在看，少个 bug 声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/凡人多烦事01/article/detail/618452

推荐阅读

相关标签