streaming-data

流式数据处理

使用现代消息代理和流处理器，构建可用于生产环境的事件流系统和实时数据管道。

何时使用此技能

在以下情况下使用此技能：

• 构建事件驱动架构和微服务通信
• 处理实时分析、监控或告警系统
• 实施数据集成管道（CDC、ETL/ELT）
• 创建日志或指标聚合系统
• 开发物联网平台或高频交易系统

核心概念

消息代理与流处理器

消息代理（Kafka、Pulsar、Redpanda）：

• 存储并分发事件流
• 提供持久性、重放能力和分区功能
• 处理生产者与消费者之间的协调

流处理器（Flink、Spark、Kafka Streams）：

• 转换和聚合流式数据
• 提供窗口、连接和有状态操作
• 执行复杂事件处理（CEP）

交付保证

至多一次：

• 消息可能丢失，不会重复
• 开销最低
• 适用于：可接受数据丢失的指标和日志

至少一次：

• 消息不会丢失，但可能重复
• 开销适中，需要幂等消费者
• 适用于：大多数应用（默认选择）

恰好一次：

• 消息不会丢失或重复
• 开销最高，需要事务处理
• 适用于：金融交易、关键状态更新

快速入门指南

第 1 步：选择消息代理

详细比较请参见 references/broker-selection.md。

快速决策：

• Apache Kafka：成熟的生态系统、企业级功能、事件溯源
• Redpanda：低延迟、兼容 Kafka、运维更简单（无需 ZooKeeper）
• Apache Pulsar：多租户、异地复制、分层存储
• RabbitMQ：传统消息队列、RPC 模式

第 2 步：选择流处理器（如有需要）

详细比较请参见 references/processor-selection.md。

快速决策：

• Apache Flink：毫秒级延迟、实时分析、CEP
• Apache Spark：批处理与流处理混合、ML 集成、分析
• Kafka Streams：嵌入微服务中，无需单独集群
• ksqlDB：用于流处理的 SQL 接口

第 3 步：实施生产者/消费者模式

选择特定语言的指南：

• TypeScript/Node.js：references/typescript-patterns.md (KafkaJS)
• Python：references/python-patterns.md (confluent-kafka-python)
• Go：references/go-patterns.md (kafka-go)
• Java/Scala：references/java-patterns.md (Apache Kafka Java Client)

常见模式

基本生产者模式

通过错误处理将事件发送到主题：

1. Create producer with broker addresses
2. Configure delivery guarantees (acks, retries, idempotence)
3. Send messages with key (for partitioning) and value
4. Handle delivery callbacks or errors
5. Flush and close producer on shutdown

基本消费者模式

使用偏移量管理处理主题中的事件：

1. Create consumer with broker addresses and group ID
2. Subscribe to topics
3. Poll for messages
4. Process each message
5. Commit offsets (auto or manual)
6. Handle errors (retry, DLQ, skip)
7. Close consumer gracefully

错误处理策略

对于生产环境系统，请实施：

• 死信队列（DLQ）：将失败的消息发送到单独的主题
• 重试逻辑：可配置重试次数，并采用退避策略
• 优雅关闭：完成处理、提交偏移量并关闭连接
• 监控：跟踪消费者滞后、错误率和吞吐量

决策框架

框架：消息代理选择

START: What are requirements?

1. Need Kafka API compatibility?
   YES → Kafka or Redpanda
   NO → Continue

2. Is multi-tenancy critical?
   YES → Apache Pulsar
   NO → Continue

3. Operational simplicity priority?
   YES → Redpanda (single binary, no ZooKeeper)
   NO → Continue

4. Mature ecosystem needed?
   YES → Apache Kafka
   NO → Redpanda (better performance)

5. Task queues (not event streams)?
   YES → RabbitMQ or message-queues skill
   NO → Kafka/Redpanda/Pulsar

框架：流处理器选择

START: What is latency requirement?

1. Millisecond-level latency needed?
   YES → Apache Flink
   NO → Continue

2. Batch + stream in same pipeline?
   YES → Apache Spark Streaming
   NO → Continue

3. Embedded in microservice?
   YES → Kafka Streams
   NO → Continue

4. SQL interface for analysts?
   YES → ksqlDB
   NO → Flink or Spark

5. Python primary language?
   YES → Spark (PySpark) or Faust
   NO → Flink (Java/Scala)

框架：语言选择

TypeScript/Node.js：

• API 网关、Web 服务、实时仪表板
• KafkaJS 库（827 个代码片段，口碑很高）

Python：

• 数据科学、ML 管道、分析
• confluent-kafka-python（192 个代码片段，评分 68.8）

Go：

• 高性能微服务、基础设施工具
• kafka-go（42 个代码片段，符合 Go 惯用风格）

Java/Scala：

• 企业应用、Kafka Streams、Flink、Spark
• Apache Kafka Java Client（683 个代码片段，评分 76.9）

高级模式

事件溯源

将状态变更存储为不可变事件。有关以下内容，请参见 references/event-sourcing.md：

• 事件存储设计模式
• 事件架构演进
• 快照策略
• 时间查询和审计跟踪

变更数据捕获（CDC）

将数据库变更捕获为事件。请参阅 references/cdc-patterns.md，了解以下内容：

• Debezium 集成（MySQL、PostgreSQL、MongoDB）
• 实时数据同步
• 微服务数据集成模式

恰好一次处理

实现事务保证。请参阅 references/exactly-once.md，了解以下内容：

• 幂等生产者
• 事务型消费者
• 端到端恰好一次管道

错误处理

生产级错误管理。请参阅 references/error-handling.md，了解以下内容：

• 死信队列模式
• 指数退避重试策略
• 背压处理
• 针对下游故障的熔断器

参考文件

决策指南

• references/broker-selection.md - Kafka、Pulsar 和 Redpanda 对比
• references/processor-selection.md - Flink、Spark 和 Kafka Streams 对比
• references/delivery-guarantees.md - 至少一次、恰好一次模式

特定语言的实现

• references/typescript-patterns.md - KafkaJS 模式（生产者、消费者、错误处理）
• references/python-patterns.md - confluent-kafka-python 模式
• references/go-patterns.md - kafka-go 模式
• references/java-patterns.md - Apache Kafka Java 客户端模式

高级主题

• references/event-sourcing.md - 事件溯源架构
• references/cdc-patterns.md - 使用 Debezium 进行变更数据捕获
• references/exactly-once.md - 事务处理
• references/error-handling.md - DLQ、重试、背压
• references/performance-tuning.md - 吞吐量优化、分区策略

验证脚本

运行这些脚本，进行无令牌验证和生成：

验证 Kafka 配置

python scripts/validate-kafka-config.py --config producer.yaml
python scripts/validate-kafka-config.py --config consumer.yaml

检查内容：代理连接、配置有效性、序列化格式

生成 Schema Registry 模板

python scripts/generate-schema.py --type avro --entity User
python scripts/generate-schema.py --type protobuf --entity Event

创建内容：供 Schema Registry 使用的 Avro/Protobuf 架构定义

吞吐量基准测试

bash scripts/benchmark-throughput.sh --broker localhost:9092 --topic test

测试内容：生产者/消费者吞吐量、延迟百分位数

代码示例

TypeScript 示例（KafkaJS）

请参阅 examples/typescript/，了解以下文件：

• basic-producer.ts - 带错误处理的简单事件生产者
• basic-consumer.ts - 手动提交偏移量的消费者
• transactional-producer.ts - 恰好一次生产者模式
• consumer-with-dlq.ts - 死信队列实现

Python 示例（confluent-kafka-python）

请参阅 examples/python/，了解以下文件：

• basic_producer.py - 带传递回调的生产者
• basic_consumer.py - 带错误处理的消费者
• async_producer.py - AsyncIO 生产者（aiokafka）
• schema_registry.py - 使用 Schema Registry 进行 Avro 序列化

Go 示例（kafka-go）

请参阅 examples/go/，了解以下文件：

• basic_producer.go - 符合 Go 惯用风格的生产者
• basic_consumer.go - 手动提交的消费者
• high_perf_consumer.go - 并发处理模式
• batch_producer.go - 批量发送消息

Java 示例（Apache Kafka）

请参阅 examples/java/，了解以下文件：

• BasicProducer.java - 启用幂等性的生产者
• BasicConsumer.java - 带错误恢复的消费者
• TransactionalProducer.java - 恰好一次事务
• StreamsAggregation.java - Kafka Streams 聚合

技术对比

消息代理对比

特性	Kafka	Pulsar	Redpanda	RabbitMQ
吞吐量	极高	高	极高	中
延迟	中	中	低	低
事件重放	是	是	是	否
多租户	手动	原生	手动	手动
运维复杂度	中	高	低	低
最适合	企业、大数据	SaaS、IoT	性能关键型场景	任务队列

流处理器对比

特性	Flink	Spark	Kafka Streams	ksqlDB
处理模型	真正的流式处理	微批处理	库	SQL 引擎
延迟	毫秒级	秒级	毫秒级	秒级
部署方式	集群	集群	嵌入式	服务端
最适合	实时分析	批处理 + 流处理	微服务	分析师

客户端库推荐

语言	库	信任评分	代码片段	使用场景
TypeScript	KafkaJS	高	827	Web 服务、API
Python	confluent-kafka-python	高（68.8）	192	数据管道、ML
Go	kafka-go	高	42	高性能服务
Java	Kafka Java Client	高（76.9）	683	企业级、Flink/Spark

相关技能

有关身份验证和安全模式，请参阅 auth-security 技能。 有关基础设施部署（Kubernetes 运算符、Terraform），请参阅 infrastructure-as-code 技能。 有关监控指标和追踪，请参阅 observability 技能。 有关 API 设计模式，请参阅 api-design-principles 技能。 有关数据架构和数据仓库，请参阅 data-architecture 技能。

故障排除

消费者延迟问题

• 检查分区数与消费者数（为实现并行处理，两者应匹配）
• 增加消费者实例，或缩短处理时间
• 使用 Kafka 消费者延迟指标进行监控

消息丢失

• 验证生产者的 acks=all 配置
• 检查代理的复制因子（>1）
• 确保消费者在处理后提交偏移量

重复消息

• 实现幂等消费者（跟踪消息 ID）
• 使用恰好一次语义（事务）
• 按至少一次投递的方式进行设计

性能瓶颈

• 增加分区数以实现并行处理
• 调整批次大小和延迟时间
• 启用压缩（GZIP、LZ4、Snappy）
• 详情请参阅 references/performance-tuning.md