生产者批次设计

批次的基本概念

● 是一组待发送的消息集合，这些消息通常会被一起发送到 Kafka 服务器，以减少网络开销。
● 批次在生产者内部创建和管理，用于将多个单独的消息打包，形成一个更大的消息单元，以便一次性发送。
● 生产者会为批次分配一定的内存空间，当批次被发送出去后，其占用的内存将被释放，以便为后续批次使用。

批次的创建和管理

触发条件：

● batch.size：
○ 这是一个配置参数，指定了一个批次所能容纳的最大字节数。当一个批次中的消息字节数累计达到 batch.size 时，该批次将被发送。
○ 例如，若 batch.size 被设置为 16384 字节，当一批次中消息的字节数达到或超过这个值时，生产者将把该批次发送到 Kafka 服务器。
● linger.ms：
○ 另一个配置参数，指定了一个批次的最长等待时间（以毫秒为单位）。
○ 即使批次未满，如果从第一个消息加入该批次开始，经过 linger.ms 的时间后，该批次也会被发送。
○ 例如，设置 linger.ms 为 100 毫秒，意味着一个批次会等待 100 毫秒，若在此期间未达到 batch.size，但时间到了，该批次也会被发送。

分区和批次的关系：

● 对于每个主题的每个分区，生产者会创建和管理相应的批次。
● 当消息被发送时，首先根据消息的 Key（若有）使用分区器确定其所属分区，然后将消息添加到该分区对应的批次中。
● 不同分区的批次是独立的，它们会根据各自的 batch.size 和 linger.ms 条件来触发发送操作。

3. 批次的内部结构

● 消息存储：
○ 一个批次内部会存储多个消息记录，这些记录包含消息的 Key、Value、时间戳等信息。
○ 存储时，消息通常以一种紧凑的形式排列，以减少空间占用。
○ 消息在批次中的存储可能会使用缓冲区，以提高存储和发送的效率。

消息结构

批次（B

批次是 Kafka 中最小的可压缩单元，包含多条记录。每个批次有如下字段：
● baseOffset: 批次中第一条消息的偏移量。
● batchLength: 批次的长度（字节数）。
● partitionLeaderEpoch: 分区领导者纪元，用于保证消费者的消费进度与领导者变更同步。
● magic: 版本号，标识使用的批次格式版本。当前版本为 2。
● crc: 循环冗余校验码，用于验证批次的完整性。
● attributes: 包含压缩类型、时间戳类型等标志位的属性字段。
● lastOffsetDelta: 批次中最后一条消息相对于 baseOffset 的增量值。
● baseTimestamp: 批次中第一条消息的时间戳。
● maxTimestamp: 批次中的最大时间戳。
● producerId: 生产者 ID，用于幂等性和事务支持。
● producerEpoch: 生产者纪元，与生产者 ID 一起使用来确保生产者的唯一性。
● baseSequence: 生产者序列号，用于幂等性。
● records: []记录列表，包含实际的消息数据。

记录（Record）

每条记录由以下部分组成：
● length: 记录的长度（变长整数）。
● attributes: 属性字段，目前未使用。
● timestampDelta: 时间戳增量（变长整数），相对于批次的 baseTimestamp。
● offsetDelta: 偏移量增量（变长整数），相对于批次的 baseOffset。
● keyLength: 键的长度（变长整数）。
● key: 消息键（字节数组）。
● valueLen: 消息体长度（变长整数）。
● value: 消息体（字节数组）。
● Headers: 消息头列表，可以携带额外的元数据。

记录头（Header）

每个消息头包括：
● headerKeyLength: 头键的长度（变长整数）。
● headerKey: 头键（字符串）。
● headerValueLength: 头值的长度（变长整数）。
● Value: 头值（字节数组）。
变长整数（varint）和变长长整数（varlong）是特殊的编码方式，旨在节省空间，只有当数值较大时才会占用更多字节。这使得 Kafka 能够有效地存储和传输消息，即使它们非常小。