【kafka系列】消费者

目录

获取消息

1. 消费者获取消息的流程逻辑分析

阶段一：消费者初始化

阶段二：分区分配与重平衡（Rebalance）

阶段三：消息拉取与处理

阶段四：偏移量提交

核心设计思想

2. 流程

关键点总结

常见参数

一、核心必填参数

二、消费者组与重平衡参数

三、消息拉取与处理参数

四、偏移量（Offset）提交参数

五、错误处理与容错参数

六、高级配置

---

获取消息

1. 消费者获取消息的流程逻辑分析

Kafka 消费者通过 消费者组（Consumer Group） 协作消费消息，核心流程分为 初始化、分区分配、消息拉取、偏移量提交 四个阶段：

---

阶段一：消费者初始化

1. 订阅 Topic

• • 消费者通过 consumer.subscribe() 订阅一个或多个 Topic。
  • 若消费者属于同一消费者组，组内消费者会均分 Topic 的分区。

1. 加入消费者组

• • 消费者启动时向 Broker 发送 JoinGroup 请求，加入消费者组。
  • 若消费者是组内第一个成员，会被选举为 Leader 消费者，负责分区分配。

---

阶段二：分区分配与重平衡（Rebalance）

1. 分区分配策略

• • Leader 消费者根据策略（如 RangeAssignor 或 RoundRobinAssignor）分配分区。
  • 分配结果通过 SyncGroup 请求同步给所有消费者。

1. 重平衡触发条件

• • 消费者加入或离开组。
  • Topic 的分区数量变化。
  • 消费者心跳超时（默认 session.timeout.ms=45s）。

---

阶段三：消息拉取与处理

1. 拉取消息

• • 消费者向分区的 Leader Broker 发送 FetchRequest，从当前偏移量（Offset）拉取消息。
  • 关键配置：

• • • max.poll.records：单次拉取最大消息数（默认 500）。
    • fetch.min.bytes：最小拉取数据量（默认 1B，优先吞吐量时可调大）。

1. 处理消息

• • 用户通过 ConsumerRecords 处理消息，需在 max.poll.interval.ms（默认 5分钟）内完成，否则触发重平衡。

---

阶段四：偏移量提交

1. 提交 Offset

• • 自动提交：由消费者线程周期性提交（enable.auto.commit=true，默认 5秒）。
  • 手动提交：用户调用 commitSync() 或 commitAsync() 精确控制。
  • Offset 存储在 Kafka 内部 Topic __consumer_offsets 中。

---

核心设计思想

• 负载均衡：通过消费者组实现分区并行消费。
• 容错性：心跳机制检测消费者存活，重平衡保障分区重新分配。
• 至少一次语义：Offset 提交后移，确保消息至少被消费一次。

---

2. 流程

---

关键点总结

1. 重平衡机制：保障消费者组动态扩展和容错。
2. Offset 管理：通过提交 Offset 实现消费进度持久化。
3. 消息拉取优化：通过 fetch.min.bytes 和 max.poll.records 平衡吞吐与延迟。
4. 超时控制：session.timeout.ms 和 max.poll.interval.ms 防止消费者僵死

常见参数

一、核心必填参数

参数名	默认值	说明
bootstrap.servers	无	Kafka 集群地址列表（逗号分隔，如 host1:9092,host2:9092 ）。
group.id	无	消费者组 ID（同一组内的消费者共享分区负载）。
key.deserializer	无	Key 的反序列化类（如 org.apache.kafka.common.serialization.StringDeserializer ）。
value.deserializer	无	Value 的反序列化类（同上）。

---

二、消费者组与重平衡参数

参数名	默认值	说明
session.timeout.ms	45000 (45秒)	消费者与 Broker 的心跳超时时间，超时触发重平衡。
heartbeat.interval.ms	3000 (3秒)	消费者发送心跳的间隔时间（需小于 session.timeout.ms 的 1/3）。
max.poll.interval.ms	300000 (5分钟)	两次 poll() 调用的最大间隔时间，超时触发重平衡。
partition.assignment.strategy	RangeAssignor	分区分配策略（如 RoundRobinAssignor 、CooperativeStickyAssignor ）。

---

三、消息拉取与处理参数

参数名	默认值	说明
fetch.min.bytes	1 (1字节)	单次拉取的最小数据量（Broker 等待足够数据后返回，提升吞吐量）。
fetch.max.bytes	52428800 (50MB)	单次拉取的最大数据量（需小于 Broker 的 message.max.bytes ）。
max.poll.records	500	单次 poll() 返回的最大消息数（避免内存溢出）。
max.partition.fetch.bytes	1048576 (1MB)	单分区单次拉取的最大数据量。

---

四、偏移量（Offset）提交参数

参数名	默认值	说明
enable.auto.commit	true	是否自动提交 Offset（建议设为 false ，手动提交确保精确控制）。
auto.commit.interval.ms	5000 (5秒)	自动提交 Offset 的时间间隔（enable.auto.commit=true 时生效）。
auto.offset.reset	latest	无初始 Offset 时的策略：<br>- earliest ：从最早消息开始。<br>- latest ：从最新消息开始。

---

五、错误处理与容错参数

参数名	默认值	说明
isolation.level	read_uncommitted	事务消息隔离级别：<br>- read_committed ：仅读取已提交的事务消息。

---

六、高级配置

参数名	默认值	说明
client.id	无	客户端标识（用于监控和日志）。
connections.max.idle.ms	540000 (9分钟)	空闲连接超时时间（Broker 主动关闭超时连接）。
request.timeout.ms	30000 (30秒)	消费者等待 Broker 响应的超时时间。