consumer C2
ISR 副本集合（In-sync Replicas）
① broker端记录每个生产者向各主题各分区的提交序号：font color=\"#ff0000\
broker B1
...
⑥ 独立线程利用已建立的TCP连接，发送到相应broker，返回消息元数据(偏移量等信息)
×
topic T1
⑧ 根据参数metadata.max.age.ms，定期给(任意?)broker发送metadata请求，更新元数据
8
向主题T1的分区0发送消息(SN=0)
0
kafka集群
2
broker B2
被打开的日志片段正在写入
② 后台创建&启动Sender线程根据bootstrap.servers与指定broker建立TCP连接
broker
1
心跳线程
如果生产者压缩消息broker会执行解压缩
重启建立连接，transaction.id = xxx
① 创建kafkaProducer实例
3
consumer C3
分区1 - 领导者副本
consumer C1
③ 向已建立连接的某broker发送metadata请求，获取集群元数据
7
幂等生产者
Java生产者与broker之间的TCP连接
① 同一分区可被多个组消费；但每个组内，只能被1个消费者消费；② 组内：消费者数 > 所有topic的所有分区数 => 存在消费者处于空闲
位移主题 __consumer_offsets
5
已关闭的日志片段等待过期
序列化器
消息集合
⑤ 调用send方法发送消息
消费者
分区1 - 追随者副本
分区0 - 追随者副本
事务生产者
kafka集群
异步拉取写入提交日志
coordinator
最后：③对应的连接会被废弃心跳线程使用④对应的连接，用于组协调请求，比如加入组、等待组分配方案、心跳请求、位移提交...用户主线程使用⑤对应的连接，用于定期请求元数据、消费消息请求
读取消息集合
6
. . .
x+3
① 构建kafkaConsumer实例
分区0 - 领导者副本
4
发送消息集合
分配Producer唯一ID (PID)
⑦(2) 给(任意??)broker发送metadata请求，更新集群元数据
消费者组 G1
③.a 请求任意broker(broker.id=-1)，创建第1个socket连接
④ 与③.c返回的broker创建socket连接broker.id=Integer.MAX_VALUE-真实的broker.id
⑦(1) 给不存在的topic发送消息，(哪个?)borker返回topic不存在
返回原PID、新的epoch
分区2 - 追随者副本
第1个socket连接
生产者
向主题T2的分区0发送消息(SN=0)
消费者组 G2
使失效
向主题T1的分区0发送消息(SN=1)
Java消费者与broker之间的TCP连接
kafka概貌
分区器
broker B3
调用send方法发送消息
① 不同分区的消息不会重复② 消息记录在磁盘，只能追加写，避免缓慢的随机I/O操作，而是顺序I/O操作
① kafka的内部主题之一② 作用：保存 Kafka 消费者的位移信息
分区2 - 领导者副本
⑩ 发送消息时，发现未连接，立即建立TCP连接
消费组group.id=xxx
⑤ 与各分区领导者副本所在broker建立TCP连接
B1-T1：分区0、分区1 B2-T1：分区1 B3-T1：分区2、分区1
④ 更新元数据后，与未建立连接span style=\
x
③.c findCoordinator请求
② 调用poll方法消费消息
① PID是对用户透明；事务id是人为指定② 事务型producer可以将消息原子地写入多个分区、不惧重启
建立TCP连接enable.idempotence = true
分配PID、单调增的epoch
⑨ 根据Producer 端参数 connections.max.idle.ms，idle时间内没有请求流过某TCP连接，broker主动关闭
根据生产者的compression.type参数启用指定类型的压缩算法
建立TCP连接enable.idempotence = true; transactional.id = xxx
③.b 发送元数据请求，获取整个集群信息
(1) 根据：xxx的哈希值%位移主题的分区数的绝对值，确定由位移主题的哪个分区保存该group数据(2) 找出该分区leader副本所在broker
生产者发布消息的客户端
异步拉取
用户主线程
