🗒️kafka 源码阅读-生产者

整体结构

• admin: 管理 Topic、Partition 的创建、删除以及扩展操作

• network: 网络通信层，处理客户端连接和事件驱动模型，包含 ScoketServer、Acceptor(主 Reactor)、Processor(从 Reactor)等组件，采用主从 Reactor多线程实现高并发。

• log: 消息持久化模块，管理日志段（LogSegment）、索引文件（.index、.timeIndex）和事务索引（.txnindex）,支持顺序读写与高效检索。

• controller: 集群元数据管理，负责 Leader 选举、Partition 分配和副本同步，KRaft 模式替代 Zookeeper 实现去中心化元数据管理

• server: Broker 服务端主逻辑，包括动态配置管理、副本同步、元数据缓存等

启动配置

server.propertis

• process.roles=broker,controller
kRaft 角色，broker处理生产/消费请求，controller 管理集群元数据（如分区分配、Leader 选举）
组合模式：单节点同时承担两种角色（适合开发环境、生产建议分离）

生产者

• ProducerConfig 中描述了生产者的相关配置，如果有些用户没有配置，就采用默认配置。

• 扩展：
• 分区器 partitionPlugin 用于自定义分区策略
• keySerializerPlugin 用于自定义消息 key 的序列化，
• valueSerializerPlugin 用于自定义消息内容的序列化

• RecordAccumulator 充当本地队列的作用，它内部维护了一个缓冲池 BuffPool，缓冲池的大小取决于buffer.memory 的配置大小，默认 32MB。生产者创建消息时，会将两次发送间隔的消息组成批次放入到 ProducerBatch 中。

• sender 线程 KafkaProducer启动时会创建一个 sender线程用于发送消息到 Borker，它会从 缓冲池中获得已就绪的 ProdcerBatch，然后将其发送到 Borker。

生产者发送消息流程

1. 消息创建与拦截处理

2. 序列化与分区选择

3. 消息追加到缓冲区（RecordAccumulator）

4. Sender 线程异步发送消息到 Broker

5. Broker 接收并持久化消息

6. 回调处理与资源释放

1.消息的创建与拦截处理

2.序列化与分区选择

分区选择

1. 如果指定了分区，则使用指定的分区

2. 自定义分区器 如果指定了
实现Partitioner 接口，来实现自定义分区策略

3. hash(如果指定了 key,则使用该方式)
发送消息时，指定了key，murmur2.hash(key)% topic分区数量。
 

1. 内置分区策略
没有指定 key的情况下，会返回一个特殊值 RecordMetadata.UNKNOWN_PARTITION（-1） ,在构建消息批次队列时，使用该值。
内置分区策略也叫粘性分区策略，该策略依赖集群元数据和每个主题分区的负载统计信息。
具体的选择分区规则如下：
1. 生成基础随机数
2. 获取当前分区负载统计快照(PartitionLoadStats)
3. 判断PartitionLoadStats 是否为空
1. PartitionLoadStats为空 （无分区负载统计信息）
根据 topic获取可用分区列表（依赖集群元数据），如果可用列表不为空，从可用分区列表随机选择一个。
如果为空则从全部分区选择一个
2. PartitionLoadStats 不为空
根据加权随机算法选择一个分区，低负载的分区被选中的几率高。
- 示例：队列大小[0,3,1] → 权重[4,1,3] → 累积[4,5,8]
• 随机数0-3 → 分区0，4 → 分区1，5-7 → 分区2

3.消息追加到缓冲区（RecordAccumulator）

消息的序列化和分区计算完成后，会追加RecordAccumulator（消息收集器）中。RecordAccumulator充当的角色是一个“队列”，间隔一段时间的消息拼凑成一个批次（ProducerBatch）再去发送可以减少网络请求的次数以提审整体的吞吐量。
RecordAccumulator 内部维护了一个缓冲池，缓冲池的大小由配置由 buffer.memory 决定（默认为 32MB）。一条消息追加到RecordAccmulator时，会先去寻找与消息分区对应的分区所对应的双端队列（如果没有则新建），再从这个双端队列的尾部获取一个 ProducerBatch(如果没有则新建)，判断当前消息是否可以写入ProucerBatch中，如果不能写入则需要创建一个新的 ProducerBatch。新建 ProducerBatch时会评估这条消息的大小是否超过了配置参数 batch.size 参数的大小，如果超过了会以消息的size从 BuffPool 申请内存空间，不过这段内存区域不会被复用。不超过则以batch.size 参数大小来创建 ProducerBatch。
 
消息处理完成后，不是立即发送的（如果配置linger.ms 为 0，则视为立即发送）。
ProducerRecord消息这里做的消息的批次处理，构建 ProducerBatch 批次的具体的结构为KV 结构，key 为分区号，value 为批次数据。，用于缓存临时数据。为了提升性能可以将消息按分区结构，组成批次数据,交给 Sender 线程将消息组成的批次数据发送给 Broker。
发送的消息会根据分区号进行批次构建或追加，消息能进入同一个 ProdcerBatch 依据两个配置 batch.size和linger.ms ，batch.size 当前的批次大小，linger.ms 等待时长。消息会首先尝试加入现有的批次中，如果批次已满（≥batchSize）追加不成功，则会申请新的批次加入到新的批次当中。将消息追加成功后，将唤醒 Sender 线程发送批次消息。
 

kafka 源码中有关于这块的描述 生产者会将两次请求发送间隔内到达的所有记录合并为一个批量请求

通常情况下，这种行为仅在系统处于高负载时发生（即记录到达速度快于发送速度）。但在某些场景下，即使处于中等负载，客户端也可能需要主动降低请求数量。该机制通过引入少量人为延迟实现——生产者不会立即发送记录，而是等待指定时长（允许其他记录加入发送队列），从而将多条记录合并发送。此机制类似于 TCP 协议中的Nagle 算法。

4. Send 线程发送消息