生产者到topic

producer发布消息

  • 写入方式

producer 采用 push 模式将消息发布到 broker,每条消息都被 append 到 patition 中,属于顺序写磁盘(顺序写磁盘效率比随机写内存要高,保障 kafka 吞吐率,关于磁盘顺序写入)。

  • 消息路由

producer 发送消息到 broker 时,会根据分区算法选择将其存储到哪一个 partition。其路由机制为:

1、 指定了 patition,则直接使用; 2、 未指定 patition 但指定 key,通过对 key 的 value 进行hash 选出一个 patition 3、 patition 和 key 都未指定,使用轮询选出一个 patition。

  • 写入流程

producer 写入消息序列图如下所示:

生产者到topic

流程说明:

1、 producer 先从 zookeeper 的 "/brokers/.../state" 节点找到该 partition 的 leader 2、 producer 将消息发送给该 leader 3、 leader 将消息写入本地 log 4、 followers 从 leader pull 消息,写入本地 log 后 leader 发送 ACK 5、 leader 收到所有 ISR 中的 replica 的 ACK 后,增加 HW(high watermark,最后 commit 的 offset) 并向 producer 发送 ACK

broker保存消息

  • 存储方式

物理上把 topic 分成一个或多个 patition(对应 server.properties 中的 num.partitions=3 配置),每个 patition 物理上对应一个文件夹(该文件夹存储该 patition 的所有消息和索引文件)

  • 存储策略

无论消息是否被消费,kafka 都会保留所有消息。有两种策略可以删除旧数据:

1、 基于时间:log.retention.hours=168 
2、 基于大小:log.retention.bytes=1073741824

Topic的创建和删除

  • 创建topic

生产者到topic

流程说明:

1、 controller 在 ZooKeeper 的 /brokers/topics 节点上注册 watcher,当 topic 被创建,则 controller 会通过 watcher 得到该 topic 的 partition/replica 分配。 2、 controller从 /brokers/ids 读取当前所有可用的 broker 列表,对于 set_p 中的每一个 partition: 2.1、 从分配给该 partition 的所有 replica(称为AR)中任选一个可用的 broker 作为新的 leader,并将AR设置为新的 ISR 2.2、 将新的 leader 和 ISR 写入 /brokers/topics/[topic]/partitions/[partition]/state 3、 controller 通过 RPC 向相关的 broker 发送 LeaderAndISRRequest。

  • 删除topic

生产者到topic

流程说明:

1、 controller 在 zooKeeper 的 /brokers/topics 节点上注册 watcher,当 topic 被删除,则 controller 会通过 watcher 得到该 topic 的 partition/replica 分配。 2、 若 delete.topic.enable=false,结束;否则 controller 注册在 /admin/delete_topics 上的 watcher 被 fired,controller 通过回调向对应的 broker 发送 StopReplicaRequest。

broker failover

kafka broker failover 序列图如下所示:

生产者到topic

流程说明:

1、 controller 在 zookeeper 的 /brokers/ids/[brokerId] 节点注册 Watcher,当 broker 宕机时 zookeeper 会 fire watcher 2、 controller 从 /brokers/ids 节点读取可用broker 3、 controller决定set_p,该集合包含宕机 broker 上的所有 partition 4、 对 set_p 中的每一个 partition 4.1、 从/brokers/topics/[topic]/partitions/[partition]/state 节点读取 ISR 4.2、 决定新 leader 4.3、 将新 leader、ISR、controller_epoch 和 leader_epoch 等信息写入 state 节点 5、 通过 RPC 向相关 broker 发送 leaderAndISRRequest 命令

controller failover

当 controller 宕机时会触发 controller failover。每个 broker 都会在 zookeeper 的 "/controller" 节点注册 watcher,当 controller 宕机时 zookeeper 中的临时节点消失,所有存活的 broker 收到 fire 的通知,每个 broker 都尝试创建新的 controller path,只有一个竞选成功并当选为 controller。

上一篇:Java如何实现消费数据隔离?


下一篇:kafka如何彻底删除topic及数据