1.启动配置

1.1 server.propertis

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

• bootstrap.servers
一组 Kafka broker 的地址，客户端会从这些地址中获取集群的元数据，以建立与整个集群的连接。

2.生产者

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

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

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

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

2.1 生产者发送消息流程

1. 消息创建与拦截处理

2. 序列化与分区选择

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

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

5. Broker 接收并持久化消息

6. 回调处理与资源释放

2.1.1 消息的创建与拦截处理

2.1.2 序列化与分区选择

2.1.3 分区选择

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
分区的选择根据消息的 Key来决定（自定义分区策略除外），
• 未指定分区的情况下，但消息有 key时，对 key 进行 hash运算，得到一个 hash 值，将 hash值与主题的分区数取模（hash(key)%分区数），这个结果就是目标的分区好。这样的做的好处是同 key 的消息会被分配到同一个分区，保证同 key 消息的顺序。
• 未指定分区且消息无 key，生产者采用轮询（Round-Robin）策略: 生产者会在可用分区中轮询发送消息，实现负载均衡。在实现中，生产者会维护一个 nextPartition，每次发送消息后递增，再与分区数取模运得到分区号。
分区可用性感知：生产者会定期获取主题的元数据，若某个分区的 Leader 不可用（如 broker 宕机），默认分区策略会跳过该分区，将消息发送到其他可用分区（避免消息丢失或发送失败）
粘性分区策略（Sticky Partioner）：减少分区切换带来的网络开销，优先将消息发送到上一次的分区（“粘性”），直到该分区满或发生故障，再切换到其他分区，该策略适用于无 key消息的场景，可减少 TCP 连接的频繁切换，提升发送效率。

2.1.4 消息追加到缓冲区（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 算法。

2.1.5 Send 线程发送消息