📚 课程目录第4章 Kafka命令行操作实战

4. Kafka 命令行操作实战：创建 Topic、生产消息、消费消息、查看消费组状态

在支付风控实时评分系统中，Kafka 命令行工具是开发与运维人员最直接的交互手段。本章将围绕四个核心操作展开：Topic 管理、消息生产、消息消费以及消费组状态监控。所有命令均基于 Kafka 3.x 版本，并假设已配置 KAFKA_HOME 环境变量。

操作前提：确保 Kafka 集群已启动，且 ZooKeeper 或 KRaft 元数据服务正常运行。以下命令默认在 Kafka 安装目录的 bin/ 下执行。

4.1 创建 Topic：定义数据通道

Topic 是 Kafka 中消息的逻辑分类。在支付风控场景中，通常为每类事件（如交易、登录、风控评分结果）创建独立的 Topic。

创建 Topic 的基本命令格式如下：

kafka-topics.sh --bootstrap-server localhost:9092 \
  --create \
  --topic payment-transactions \
  --partitions 3 \
  --replication-factor 2

参数	说明	支付风控推荐值
`--bootstrap-server`	Kafka 集群地址	localhost:9092（开发环境）
`--topic`	Topic 名称	按业务命名，如 `risk-score-result`
`--partitions`	分区数，决定并行度	3~6（根据吞吐量调整）
`--replication-factor`	副本因子，保障高可用	2~3（生产环境至少 2）

  风控最佳实践：支付风控场景中，建议将 --partitions 设置为消费者组最大并行度的 2~3 倍，以应对流量突发。例如，若风控评分服务部署 4 个实例，分区数可设为 8。

验证 Topic 是否创建成功：

kafka-topics.sh --bootstrap-server localhost:9092 --list

查看 Topic 的详细信息（分区、副本、ISR 等）：

kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic payment-transactions

4.2 生产消息：模拟风控事件

使用 kafka-console-producer.sh 向 Topic 发送消息。在支付风控场景中，消息通常为 JSON 格式，包含交易 ID、金额、时间戳等字段。

启动生产者并发送消息：

kafka-console-producer.sh --bootstrap-server localhost:9092 \
  --topic payment-transactions \
  --property "parse.key=true" \
  --property "key.separator=:"

输入消息示例（key:value 格式）：

txn_001:{"txnId":"txn_001","amount":1250.00,"currency":"CNY","timestamp":1710000000}
txn_002:{"txnId":"txn_002","amount":8900.00,"currency":"CNY","timestamp":1710000005}

注意：在生产环境中，消息 key 用于决定消息发往哪个分区。若 key 为 null，Kafka 使用轮询（Round-Robin）策略分发；若 key 非空，则对 key 进行哈希取模，相同 key 的消息始终进入同一分区，保证顺序性。

4.3 消费消息：实时处理风控数据

使用 kafka-console-consumer.sh 消费消息。消费组（Consumer Group）是 Kafka 实现水平扩展和容错的核心机制。

启动消费者并消费消息：

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic payment-transactions \
  --group risk-scoring-group \
  --from-beginning \
  --property "print.key=true" \
  --property "print.value=true"

参数	说明	使用场景
`--group`	指定消费组名称	风控评分服务实例共享同一组名
`--from-beginning`	从 Topic 最早的消息开始消费	测试环境或数据回溯
`--property print.key=true`	打印消息 key	调试分区分配策略

  风控消费模式：在支付风控系统中，每个消费者实例处理一个或多个分区。当实例数增加时，Kafka 自动重新平衡（Rebalance）分区分配，实现负载均衡。建议消费者实例数不超过分区数，避免资源浪费。

4.4 查看消费组状态：监控消费进度

消费组状态是风控系统健康度的关键指标。使用 kafka-consumer-groups.sh 查看消费组的偏移量（Offset）和延迟（Lag）。

查看消费组列表：

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

查看指定消费组的详细信息：

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group risk-scoring-group \
  --describe

输出示例：

GROUP                TOPIC                 PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG   CONSUMER-ID     HOST
risk-scoring-group   payment-transactions  0          150             200             50    consumer-1      /192.168.1.10
risk-scoring-group   payment-transactions  1          180             200             20    consumer-1      /192.168.1.10
risk-scoring-group   payment-transactions  2          190             200             10    consumer-2      /192.168.1.11
risk-scoring-group   payment-transactions  3          195             200              5    consumer-2      /192.168.1.11

字段	含义	风控监控阈值
`CURRENT-OFFSET`	消费者已消费的最新偏移量	—
`LOG-END-OFFSET`	Topic 分区的最新消息偏移量	—
`LAG`	未消费的消息数（LOG-END-OFFSET - CURRENT-OFFSET）	建议 < 1000，超过则触发告警
`CONSUMER-ID`	消费者实例的唯一标识	用于定位异常实例

风控告警规则：当任意分区的 LAG 持续增长超过阈值（如 1000）时，表明消费者处理能力不足或出现故障。此时应检查消费者实例状态、网络延迟或风控评分逻辑性能瓶颈。

4.5 重置消费组偏移量：应对数据回溯

在风控模型迭代或数据修复场景中，可能需要重置消费组偏移量以重新消费历史数据。

重置到最早偏移量：

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group risk-scoring-group \
  --topic payment-transactions \
  --reset-offsets --to-earliest --execute

重置到指定时间戳：

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group risk-scoring-group \
  --topic payment-transactions \
  --reset-offsets --to-datetime 2025-01-01T00:00:00.000 --execute

  注意：重置偏移量前，必须停止消费组的所有消费者实例，否则操作会失败。重置后，消费者将从指定位置重新消费消息，适用于风控规则回测或数据补录。

4.6 本章小结

通过本章实战，你已掌握 Kafka 命令行操作的核心技能：

创建 Topic：指定分区数和副本因子，为支付风控事件定义数据通道。
生产消息：使用 key-value 格式发送 JSON 风控事件，确保相同 key 的消息顺序性。
消费消息：通过消费组实现水平扩展，每个分区只被组内一个消费者处理。
查看消费组状态：监控 LAG 指标，及时发现消费瓶颈。
重置偏移量：在模型迭代或数据修复时灵活回溯历史数据。

这些命令行操作是后续构建 Kafka Streams 风控评分拓扑的基础。在生产环境中，建议将常用命令封装为脚本，并集成到监控告警系统中。