producer发布消息

• 写入方式

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

• 消息路由

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

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

• 写入流程

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

流程说明：

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

流程说明：

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

流程说明：

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 序列图如下所示：

流程说明：

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。