为什么不用现成的 ConcurrentHashMap
很多人写 Producer 封装会用ConcurrentHashMap<TopicPartition, Queue<ProducerRecord>>——每来一条消息塞进去,后台线程消费。功能上没问题,但性能差一个数量级。
Kafka 没有这么做。它花了很大力气设计自己的数据结构,核心目标三个:
- 攒批:多条消息合并成一个网络请求,而不是每条一次请求
- 内存复用:避免频繁 new/delete ByteBuffer,减少 GC
- 背压:Producer 不能无限快地往 Broker 塞,需要限流
01 Deque:为什么不是 Queue
每个分区的批次队列用的是Deque<ProducerBatch>,不是普通的Queue。原因是读写两端不同:
- 写端(right):业务线程把新批次加到队尾,正在攒批
- 读端(left):Sender 线程从队头拉 ready 批次发出去
- 失败重传:一个批次发出去但 ACK 失败,要重新入队——插到队头,不是队尾
为什么重传要插队头?因为这个批次已经写满了,是 ready 的,应该尽快重试。如果插到队尾,后面攒的新批次会排在它前面——新批次还没攒满,要等很久才发,导致重传延迟。
Deque 的addFirst()和pollLast()都是 O(1),两端操作互不影响。普通 Queue 做不到这一点。
02 BufferPool:为什么要自己实现内存池
如果每个批次都ByteBuffer.allocate(16KB),发完了ByteBuffer不引用了等 GC,会发生什么?
- GC 压力:每秒几万个 ByteBuffer 对象,Young GC 频繁,STW 影响延迟
- 内存碎片:16KB 的块反复分配释放,堆碎片化
- 分配开销:Java NIO 的 allocate 本身有开销
Kafka 的做法是自己做一个池:
buffer.memory = 33554432(32MB) batch.size = 16384(16KB)32MB 总内存切成 2048 个 16KB 的块,放在一个free: Deque<ByteBuffer>里。
分配时:free 队列里有空块就直接pollFirst(),没有就 new 一个。如果总内存超了(32MB 用完了),业务线程阻塞在memoryAvailable.await(),等 ACK 回来释放块。
释放时:ACK 回来,批次从 incomplete 移除,ByteBuffer 放回 free 队列,signalAll()唤醒等待者。
整个过程没有 ByteBuffer 被 GC——32MB 的块循环使用,长期运行内存占用稳定。这就是为什么 Kafka Producer 跑一天 OOM 问题很少(除非配置错了)。
线上踩过的坑:buffer.memory 设得太小。50 个分区每个攒一个 16KB 批次就是 800KB,看起来不多,但如果 Broker 响应慢(比如跨机房),批次都在 incomplete 里不释放,32MB 很快用完,业务线程阻塞 60 秒后抛异常。症状就是"偶发性 send 超时"——调大 buffer.memory 或排查 Broker 网络。
03 InFlightRequests:为什么要限制在途请求数
默认max.in.flight.requests.per.connection = 5。意思是:每个 TCP 连接上,最多有 5 个未收到 ACK 的请求同时在飞。
为什么要有这个限制?
不限制会怎样?Producer 可以同时发 1000 个请求给 Broker,Broker 处理不过来,内存被打爆,最后 OOM 崩溃。Producer 不是"越快越好"——它要和 Broker 的处理能力匹配。
限制了会怎样?第 6 个请求必须等前面某个请求 ACK 回来。这天然形成了背压——Producer 的发送速度不会超过 Broker 的处理速度。
还有个副作用:acks=0或acks=1时,消息乱序风险。如果第 1 个请求慢、第 2 个请求快,第 2 个先到 Broker,同分区的两条消息顺序就反了。要严格保证顺序,设max.in.flight.requests.per.connection=1——但性能会掉一半以上。大部分业务接受弱顺序(同一用户的消息不要求严格有序),保留 5 这个默认值。
04 ConcurrentMap:分区之间无锁
batches是ConcurrentHashMap<TopicPartition, Deque<ProducerBatch>>。
为什么不用普通 HashMap?因为多个业务线程同时 send 不同分区时,ConcurrentHashMap 只锁桶,两个线程写不同分区完全并行。如果用 synchronized HashMap,所有 send 都串行化——多线程 Producer 等于单线程。
同一个分区的 append 还是要加锁的(synchronized on the deque),但不同分区之间零竞争。这就是为什么 Kafka Producer 在多线程场景下吞吐能线性提升——分区数越多,并发度越高。
05 写在最后
Kafka Producer 的高性能不是某个魔法参数,而是数据结构的选择:Deque 解决攒批和重传,BufferPool 解决 GC,InFlightRequests 解决背压,ConcurrentHashMap 解决多线程并发。理解了这四个结构,就理解了为什么 batch.size、linger.ms、buffer.memory、max.in.flight 这些参数是这样设计的——每一个都对应一个数据结构的行为。