Kafka Streams、Connect 与生态

学习目标

Kafka 不只是消息中间件，还包含流处理、数据集成和跨集群复制生态。本章覆盖：

• Kafka Streams：在应用内做流计算。
• Kafka Connect：标准化数据采集和落地。
• Schema Registry：治理事件结构。
• MirrorMaker 2：跨集群复制。

Kafka Streams

Kafka Streams 是 Kafka 官方 Java 流处理库。它不是独立集群，而是嵌入你的应用进程。

适合：

• 实时计数。
• 订单状态聚合。
• 风控规则。
• 双流 join。
• 窗口统计。

不适合：

• 超大规模复杂 SQL 分析，优先考虑 Flink/Spark。
• 多语言团队强依赖非 Java 技术栈。

Streams 核心概念

Streams 示例：统计订单金额

伪代码：

StreamsBuilderbuilder=newStreamsBuilder();KStream<String,OrderEvent>orders=builder.stream("order-events");orders.filter((key,value)->"PAID".equals(value.status())).groupByKey().aggregate(()->BigDecimal.ZERO,(key,value,total)->total.add(value.amount()),Materialized.as("order-amount-store")).toStream().to("order-amount-summary");

这段逻辑表达的是：

读取 order-events -> 只保留 PAID -> 按订单 key 聚合金额 -> 输出汇总 topic

Kafka Connect

Kafka Connect 用于把外部系统与 Kafka 连接起来，减少每个团队重复写采集和落地代码。

两类 Connector：

典型链路：

MySQL binlog -> Debezium Source Connector -> Kafka -> Elasticsearch Sink Connector

Connect 运行模式

生产建议使用 Distributed 模式，因为它支持：

• worker 扩容。
• connector task 分配。
• 配置存储在 Kafka topic。
• 故障后自动恢复。

Schema Registry

随着 topic 被多个系统订阅，事件结构必须治理。否则一个字段改名就可能导致多个消费者失败。

常见格式：

• JSON：简单直观，但缺少强约束。
• Avro：常配合 Schema Registry，适合数据平台。
• Protobuf：跨语言强类型，体积较小。

Schema 演进规则：

推荐事件兼容策略：

只新增可选字段，不随意删除或改名；破坏性变更使用新 topic 或 eventVersion。

MirrorMaker 2

MirrorMaker 2 用于 Kafka 集群间复制。

场景：

• 同城双活读取。
• 异地灾备。
• 机房迁移。
• 云上云下数据同步。

复制链路：

source cluster topic -> MM2 connector -> target cluster topic

注意事项：

• 跨集群复制有延迟，不是强一致。
• topic 命名可能带 source cluster alias。
• offset 同步需要额外配置和验证。
• 灾备切换前要明确 RPO/RTO。

生态选型

实操建议

学习阶段：

1. 先掌握普通 producer/consumer。
2. 再学习 Connect，用现成 connector 接入文件或数据库。
3. 再学习 Streams，理解流、表、窗口和状态。
4. 最后学习 Schema 和跨集群复制。

生产阶段：

• 所有跨团队共享 topic 必须有 Schema 文档。
• Connector 任务必须有错误队列、重试、监控和告警。
• Streams 应用必须监控 lag、state store 大小、处理延迟。
• 跨集群复制必须定期演练切换。

学习目标

本章面向生产环境，解决 Kafka 上线后怎么治理：

• 集群部署和滚动升级。
• 监控指标和告警阈值。
• 安全认证与权限。
• 常见故障排查。
• 故障演练和运维清单。

生产集群基本建议

Topic 治理规范

每个生产 topic 都应该登记这些信息：

监控指标

Broker 指标

Consumer 指标

Producer 指标

告警建议

安全

Kafka 安全包含三层：

1. 传输加密：SSL/TLS。
2. 身份认证：SASL/PLAIN、SCRAM、Kerberos、mTLS。
3. 授权：ACL。

ACL 示例：

kafka-acls --bootstrap-server kafka-1:9092\--add\--allow-principal User:order-service\--operationWrite\--topicorder-events

给消费组授权：

kafka-acls --bootstrap-server kafka-1:9092\--add\--allow-principal User:inventory-service\--operationRead\--topicorder-events\--groupinventory-service

安全原则：

• 按服务账号授权，不共享账号。
• 生产者只给写权限。
• 消费者只给读 topic 和读 group 权限。
• 禁止业务服务拥有集群级管理权限。
• 密钥定期轮换。

常见故障排查

消费堆积

排查顺序：

1. 查看 lag 是否持续增长。
2. 看消费者日志是否报错或重试。
3. 看单条处理耗时和批处理耗时。
4. 看下游数据库、缓存、HTTP 是否慢。
5. 看消费者实例数是否小于 partition 数。
6. 看是否频繁 rebalance。

临时处理：

• 扩容消费者实例。
• 降低每条消息处理成本。
• 暂停非核心消费者。
• 将坏消息转入 DLT。
• 对下游做限流保护。

ISR 缩小

排查顺序：

1. Broker 是否宕机。
2. 网络是否抖动。
3. 磁盘 IO 是否高。
4. follower 是否 GC 或 CPU 飙高。
5. topic 写入是否突增。

风险：acks=all且 ISR 小于min.insync.replicas时，producer 会写入失败。这是保护机制，不应该直接降低可靠性配置掩盖问题。

Producer 超时

常见原因：

• broker 请求队列满。
• topic leader 不可用。
• ISR 不足导致无法满足acks=all。
• 网络延迟高。
• 生产端 buffer 满。

排查指标：

• producer request latency。
• producer buffer available bytes。
• broker request handler idle。
• under replicated partitions。

滚动升级

升级前：

• 备份配置。
• 确认 controller 和 broker 状态健康。
• 检查 under replicated partitions 为 0。
• 阅读版本兼容说明。
• 先升级非核心或测试集群。

升级中：

停止一台 broker -> 升级 -> 启动 -> 等 ISR 恢复 -> 升级下一台

升级后：

• 检查 controller 数。
• 检查 ISR。
• 检查 producer error rate。
• 检查 consumer lag。
• 检查日志异常。

故障演练