3. Kafka 核心概念：Topic、Partition、Producer、Consumer、Broker、Zookeeper 的作用与关系

在金融图谱回放场景中，Kafka 作为事件流平台，承担着交易事件、图谱变更日志的可靠传输与存储。理解其六大核心概念是搭建回放验证管道的前提。下面逐一拆解每个角色的职责与协作关系。

3.1 六大核心概念速览

3.2 核心关系与架构图

下图展示了 Kafka 集群中 Broker、Topic、Partition、Producer、Consumer 与 Zookeeper 之间的静态拓扑关系。Producer 将消息发往 Topic 的指定 Partition，Consumer Group 内的实例各自负责消费不同 Partition，Zookeeper 维护集群元数据。

3.3 各组件详细职责

3.3.1 Broker

• 物理节点：每个 Broker 是一个独立的 Kafka 进程，通常部署在一台服务器上。
• 集群角色：多个 Broker 组成集群，其中一个被选举为 Controller，负责管理分区 Leader 的分配与重平衡。
• 存储与复制：每个 Partition 在多个 Broker 上存有副本（Leader/Follower），保证数据不丢失。

3.3.2 Topic

• 逻辑分类：Topic 是消息的类别名称，类似数据库中的表。
• 金融场景：可创建 trade-events、graph-changes、risk-alerts 等 Topic，分别存储不同类型的事件。
• 多订阅：一个 Topic 可被多个 Consumer Group 独立消费，互不影响。

3.3.3 Partition

• 物理分片：每个 Topic 可划分为多个 Partition，每个 Partition 是一个有序、不可变的日志文件。
• 顺序保证：同一 Partition 内的消息按写入顺序存储，Consumer 按顺序读取。
• 并行度：Partition 数量决定了 Consumer Group 内的最大并行消费数。
• 金融回放关键：按实体 ID（如股票代码）哈希到固定 Partition，保证同一实体的图谱变更顺序回放。

3.3.4 Producer

• 消息发布者：负责将消息发送到指定的 Topic。
• 分区策略：可指定 key 进行哈希分区，也可轮询或自定义分区器。
• 确认机制：acks 参数控制消息持久化确认级别（0、1、all），金融场景建议 acks=all 保证不丢数据。

3.3.5 Consumer

• 消息订阅者：从 Topic 的 Partition 拉取消息进行处理。
• Consumer Group：多个 Consumer 实例组成一个 Group，共同消费一个 Topic，每个 Partition 只能被 Group 内的一个 Consumer 消费。
• 偏移量管理：Consumer 自动或手动提交消费偏移量（offset），记录已处理的位置。
• 回放验证：回放引擎作为 Consumer，从指定 offset 开始重演历史事件，验证图谱状态。

3.3.6 Zookeeper

• 元数据存储：记录 Broker 列表、Topic 配置、Partition 副本分配等元数据。
• Controller 选举：当 Controller Broker 宕机时，Zookeeper 触发新选举，保证集群自愈。
• 集群成员管理：Broker 启动时在 Zookeeper 注册临时节点，宕机后自动移除。
• 注意：Kafka 2.8+ 开始支持 KRaft 模式逐步替代 Zookeeper，但生产环境仍广泛使用 ZK。

3.4 消息流转流程

下图展示了一条图谱变更事件从产生到被回放引擎消费的完整生命周期，突出 Producer、Broker、Partition、Consumer 之间的数据流。

3.5 关键配置示例

以下是一个 Producer 发送图谱变更事件的 Java 代码片段，展示如何指定 Topic、key 与确认级别。

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3: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", 3);
props.put("enable.idempotence", true);

Producer<String, String> producer = new KafkaProducer<>(props);

// 以 entityId 作为 key，保证同一实体的图谱变更顺序
String entityId = "stock_600519";
String graphEvent = "{\"type\":\"NODE_UPDATE\",\"entity\":\"stock_600519\",\"timestamp\":1700000000}";

ProducerRecord<String, String> record = new ProducerRecord<>("graph-changes", entityId, graphEvent);
producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        System.out.println("写入成功: topic=" + metadata.topic()
            + ", partition=" + metadata.partition()
            + ", offset=" + metadata.offset());
    } else {
        System.err.println("写入失败: " + exception.getMessage());
    }
});
producer.close();

3.6 组件关系总结