2. Kafka 基础概念与集群搭建：Topic、Partition、Broker、Producer、Consumer 核心概念；单机与集群环境搭建步骤

2.1 核心概念速览

在信用评分场景中，Kafka 作为事件总线，承载着用户授权、交易流水、模型评分、决策结果等关键消息。理解以下 5 个核心概念是搭建链路的基础。

概念	角色	信用评分场景类比
Broker	Kafka 服务器节点，负责存储和转发消息	评分系统的消息中转站，每个 Broker 可类比为一个数据中心机柜
Topic	消息的逻辑分类，类似数据库中的表	credit-score-input（原始数据）、credit-score-result（评分结果）
Partition	Topic 的物理分片，实现并行读写与水平扩展	按用户 ID 哈希分区，保证同一用户的评分请求有序处理
Producer	消息生产者，向 Topic 发送数据	风控前置系统、APP 端上报用户行为数据的服务
Consumer	消息消费者，从 Topic 拉取数据进行处理	评分引擎、规则引擎、结果存储服务

2.2 Topic 与 Partition 的物理模型

Topic 是逻辑容器，Partition 是物理存储单元。每个 Partition 对应一个日志文件（.log），消息追加写入，Consumer 通过 offset 标记读取位置。

2.3 Producer 与 Consumer 的工作机制

• Producer 发送流程： 序列化 → 分区器（Partitioner）选择目标 Partition → 批量打包 → 发送到 Leader Broker → 等待 ack 确认。
• Consumer 消费流程： 订阅 Topic → 加入消费组（Group）→ 分区再均衡（Rebalance）→ 拉取消息 → 业务处理 → 手动提交 offset。

2.4 单机环境搭建步骤（开发/测试用）

以下步骤基于 Kafka 3.x 版本，适用于本地快速验证信用评分消息链路。

1. 下载解压： 从 Apache 官网下载 kafka_2.13-3.5.0.tgz，解压至 /opt/kafka。
2. 启动 ZooKeeper（Kafka 3.x 可选内置 KRaft 模式，但生产仍推荐 ZK）：

# 启动 ZooKeeper（单机）
bin/zookeeper-server-start.sh config/zookeeper.properties &

3. 修改 Broker 配置： 编辑 config/server.properties，关键参数如下：

# 单机配置核心参数
broker.id=0
listeners=PLAINTEXT://localhost:9092
log.dirs=/tmp/kafka-logs
num.partitions=3
default.replication.factor=1
offsets.topic.replication.factor=1

4. 启动 Kafka Broker：

bin/kafka-server-start.sh config/server.properties &

5. 验证： 创建测试 Topic 并发送/消费消息。

# 创建 Topic
bin/kafka-topics.sh --create --topic credit-score-test \
  --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

# 查看 Topic 详情
bin/kafka-topics.sh --describe --topic credit-score-test \
  --bootstrap-server localhost:9092

2.5 集群环境搭建步骤（生产推荐 3 节点）

信用评分生产环境至少需要 3 个 Broker 节点，保证副本冗余与高可用。以下以 3 台服务器（kafka1, kafka2, kafka3）为例。

节点	Broker ID	IP 示例	监听端口	日志目录
kafka1	1	192.168.1.101	9092	/data/kafka/logs
kafka2	2	192.168.1.102	9092	/data/kafka/logs
kafka3	3	192.168.1.103	9092	/data/kafka/logs

1. 所有节点安装 JDK 11+ 并解压 Kafka。
2. 配置 ZooKeeper 集群（3 节点 ZK 或使用 Kafka 内置 KRaft）。 以下以 ZK 为例：

# 每个节点 zoo.cfg 配置
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper/data
clientPort=2181
server.1=192.168.1.101:2888:3888
server.2=192.168.1.102:2888:3888
server.3=192.168.1.103:2888:3888

3. 配置每个 Broker 的 server.properties：

# kafka1 配置示例
broker.id=1
listeners=PLAINTEXT://192.168.1.101:9092
advertised.listeners=PLAINTEXT://192.168.1.101:9092
log.dirs=/data/kafka/logs
num.partitions=6
default.replication.factor=3
offsets.topic.replication.factor=3
zookeeper.connect=192.168.1.101:2181,192.168.1.102:2181,192.168.1.103:2181

4. 依次启动各节点 Broker：

# 每个节点执行
bin/kafka-server-start.sh config/server.properties &

5. 验证集群状态：

# 查看所有 Broker
bin/kafka-broker-api-versions.sh --bootstrap-server 192.168.1.101:9092

# 创建带副本的 Topic
bin/kafka-topics.sh --create --topic credit-score-input \
  --bootstrap-server 192.168.1.101:9092 \
  --partitions 6 --replication-factor 3

# 查看分区副本分布
bin/kafka-topics.sh --describe --topic credit-score-input \
  --bootstrap-server 192.168.1.101:9092

2.6 集群验证与常用运维命令

操作	命令示例	说明
查看所有 Topic	kafka-topics.sh --list --bootstrap-server localhost:9092	列出集群中所有 Topic
查看消费者组	kafka-consumer-groups.sh --list --bootstrap-server localhost:9092	查看所有消费组
查看消费组偏移	kafka-consumer-groups.sh --describe --group credit-score-group --bootstrap-server localhost:9092	查看每个 Partition 的当前 offset 与 lag
生产消息测试	kafka-console-producer.sh --topic credit-score-input --bootstrap-server localhost:9092	手动输入消息验证
消费消息测试	kafka-console-consumer.sh --topic credit-score-input --bootstrap-server localhost:9092 --from-beginning	从头消费验证