【kafka系列】生产者

news2025/2/20 4:30:08

发送流程

1. 流程逻辑分析

阶段一：主线程处理

阶段二：Sender 线程异步发送

核心设计思想

2. 流程

关键点总结

重要参数

一、核心必填参数

二、可靠性相关参数

三、性能优化参数

四、高级配置

五、安全性配置（可选）

六、错误处理与监控

典型配置示例

关键注意事项

发送流程

序列化与分区：消息通过Partitioner选择目标分区（默认轮询或哈希），序列化后加入RecordAccumulator缓冲区。
批次合并：Sender线程将同一分区的消息合并为ProducerBatch，减少网络请求（源码见Sender.run()方法）。
发送至Broker：通过NetworkClient异步发送，Broker的LogAppendTime处理写入请求。
ACK机制：根据acks配置（0/1/all）等待Broker确认，通过Metadata类更新分区元数据

1. 流程逻辑分析

Kafka 生产者发送消息的核心流程分为 主线程处理 和 Sender 线程异步发送 两个阶段，具体步骤如下：

阶段一：主线程处理

创建 ProducerRecord

- 用户调用 producer.send(ProducerRecord)，指定 Topic、Key、Value 和可选的分区或时间戳。

选择分区（Partition）

- 若未指定分区，根据以下规则选择：

- - 有 Key：对 Key 哈希取模（hash(key) % 分区数），确保相同 Key 的消息进入同一分区。
  - 无 Key：默认使用粘性分区策略（Sticky Partitioning，Kafka 2.4+），在批次填满或超时前发送到同一分区，提升性能。

序列化（Serialize）

- 使用配置的 key.serializer 和 value.serializer 对 Key 和 Value 序列化（如 StringSerializer、ByteArraySerializer）。

追加到缓冲区（RecordAccumulator）

- 将消息按 Topic-Partition 分组，存入 RecordAccumulator 的批次（Batch）中。
- 批次策略：

- - batch.size：批次大小阈值（默认 16KB），达到阈值立即发送。
  - linger.ms：批次等待时间（默认 0ms），超时后发送未满批次。

阶段二：Sender 线程异步发送

Sender 线程拉取批次

- Sender 线程定期检查缓冲区，将满足条件的批次（已满或超时）封装为 ProducerRequest。

构建请求并发送到 Broker

- 根据分区的 Leader 副本所在 Broker，将请求发送到对应的节点。
- 关键配置：

- - acks：控制消息持久化确认级别：

- - - 0：不等待确认（可能丢失数据）。
    - 1：等待 Leader 确认（默认）。
    - all：等待所有 ISR 副本确认（最高可靠性）。

- - max.in.flight.requests.per.connection：控制单个 Broker 的未确认请求数（默认 5）。

处理 Broker 响应

- 成功：触发用户设置的 Callback 回调，并释放批次内存。
- 失败：

- - 可重试错误（如网络抖动、Leader 切换）：根据 retries（默认 0）和 retry.backoff.ms（默认 100ms）重试。
  - 不可重试错误（如消息过大）：直接触发回调并抛出异常。

核心设计思想

异步批处理：通过缓冲区合并小消息，减少网络 I/O 次数。
零拷贝优化：使用 sendfile 系统调用提升网络传输效率。
高可靠性：通过重试机制和 acks=all 确保消息不丢失。

2. 流程

关键点总结

分区选择：优先使用 Key 哈希或粘性分区策略，保证消息顺序性和吞吐量。
批次优化：通过 batch.size 和 linger.ms 平衡延迟与吞吐。
可靠性保障：通过 acks 和 retries 配置确保消息持久化。
异步处理：主线程与 Sender 线程解耦，避免阻塞用户逻辑。

重要参数

以下是 Kafka 生产者（Producer）在日常开发中的 常见配置参数 及其作用，按功能分类整理成表格：

一、核心必填参数

参数名	默认值	说明
`bootstrap.servers`	无	Kafka 集群地址列表（逗号分隔，如 `host1:9092,host2:9092` ）。
`key.serializer`	无	Key 的序列化类（如 `org.apache.kafka.common.serialization.StringSerializer` ）。
`value.serializer`	无	Value 的序列化类（同上）。

二、可靠性相关参数

参数名	默认值	说明
`acks`	`1`	消息持久化确认机制： `0`：不等待确认（可能丢失数据）。 `1`：等待 Leader 确认（默认）。`all`：等待所有 ISR 副本确认（最高可靠性）。
`retries`	`0`	发送失败后的重试次数（建议设为 `Integer.MAX_VALUE` 配合 `delivery.timeout.ms` ）。
`enable.idempotence`	`false`	是否启用幂等性（`true`时保证消息不重复，需配合 `acks=all` 和 `retries>0`）。
`max.in.flight.requests.per.connection`	`5`	单个 Broker 的未确认请求数。若启用幂等性，建议设为 `1` 以保证顺序。

三、性能优化参数

参数名	默认值	说明
`linger.ms`	`0`	消息在缓冲区等待时间（毫秒），增大可提升吞吐量（但增加延迟）。
`batch.size`	`16384` (16KB)	单个批次的大小阈值，达到阈值后立即发送。
`buffer.memory`	`33554432` (32MB)	生产者缓冲区的总内存大小。
`compression.type`	`none`	消息压缩算法（`gzip` 、`snappy` 、`lz4` 、`zstd` ），减少网络带宽占用。

四、高级配置

参数名	默认值	说明
`request.timeout.ms`	`30000` (30秒)	生产者等待 Broker 响应的超时时间。
`max.block.ms`	`60000` (60秒)	生产者缓冲区满或元数据不可用时的阻塞时间（超时抛异常）。
`partitioner.class`	默认轮询/哈希策略	自定义分区策略（实现 `Partitioner` 接口）。

五、安全性配置（可选）

参数名	默认值	说明
`security.protocol`	`PLAINTEXT`	安全协议（如 `SSL` 、`SASL_SSL` ）。
`ssl.keystore.location`	无	SSL 证书路径（客户端认证时需配置）。
`sasl.mechanism`	无	SASL 认证机制（如 `PLAIN` 、`SCRAM-SHA-256` ）。

六、错误处理与监控

参数名

默认值

说明

interceptor.classes

无

生产者拦截器（实现 ProducerInterceptor

接口），用于监控或修改消息。

metrics.sample.window.ms

30000

(30秒)

性能指标采样窗口时间。

典型配置示例

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "all");
props.put("retries", 10);
props.put("linger.ms", 20);
props.put("batch.size", 32768);
props.put("compression.type", "snappy");
props.put("enable.idempotence", "true");