4、实时数据流与契约：Kafka Topic契约化、Schema Registry集成、Avro/Protobuf契约定义
实时数据流的质量保障，说白了就是要在数据还在管道里飞奔的时候，就把规矩立好。
我见过太多团队，Kafka 集群跑得飞起，但消费端突然报错——字段类型对不上、枚举值变了、甚至字段直接被删了。这种问题，靠事后告警根本来不及。所以，数据契约在实时流里的落地，核心就是三件事：Topic 契约化、Schema Registry 集成、序列化格式选型。
4.1 Kafka Topic 契约化：不只是建个Topic
很多人觉得 Topic 就是个消息通道，建好就行。其实不然。Topic 本身就是一个隐形的契约载体。
我习惯把 Topic 的命名、分区数、副本数、消息保留策略，都写进契约里。为什么？因为生产端和消费端一旦约定好，谁都不能随意改。
| 契约项 | 说明 | 常见坑 |
|---|---|---|
| Topic 命名规范 | 如：{domain}.{entity}.{event_type}.v{version} | 命名混乱导致运维灾难 |
| 分区数 | 根据吞吐量预估，写死不可变 | 后期改分区数会打乱顺序保证 |
| 副本因子 | 至少 2，建议 3 | 单副本丢数据没人赔 |
| 保留策略 | 按时间或大小，明确清理规则 | 保留太久磁盘爆，太短回溯难 |
| 生产者 ACK 模式 | all / 1 / 0，必须约定 | all 保证最强但延迟高 |
我曾经遇到一个项目，生产端偷偷把 Topic 的保留时间从 7 天改成了 1 天，结果消费端凌晨批量回溯数据时，发现消息已经没了。嗯，从那以后，我把 Topic 配置写进了 CI/CD 的契约检查里，改配置必须走审批。
4.2 Schema Registry 集成：统一“方言”
Kafka 本身不关心消息体里是什么。它只认字节数组。这就带来了一个问题：生产端写了个 JSON，消费端怎么知道字段结构？
Schema Registry 就是来解决这个问题的。它像一个中央字典，所有生产者和消费者都去那里查“方言”的定义。
集成 Schema Registry 后，数据契约就变成了一个可注册、可版本化、可兼容性检查的实体。每次生产端发消息，都会带上 Schema ID；消费端拿到 ID 后，去 Registry 拉取 Schema，反序列化。
这里有个关键点：兼容性策略。我见过最惨的案例，是有人把字段类型从 int 改成了 string，结果所有老消费者全部崩溃。为什么？因为没配兼容性检查。
| 兼容性策略 | 含义 | 适用场景 |
|---|---|---|
| BACKWARD | 新 Schema 可以读旧数据 | 消费者先升级 |
| FORWARD | 旧 Schema 可以读新数据 | 生产者先升级 |
| FULL | 双向兼容 | 最安全，但限制多 |
| NONE | 不检查 | 别用，除非你想背锅 |
4.3 Avro vs Protobuf：契约定义怎么选？
选序列化格式，其实就是选契约的“语法”。目前主流就两个：Avro 和 Protobuf。JSON 虽然也能用，但缺 Schema 强校验，我一般不推荐在生产流里用。
我个人习惯：数据湖场景用 Avro，微服务间 RPC 用 Protobuf。为什么？
- Avro：Schema 和数据可以一起存，适合写进 HDFS 或 S3。而且它天生支持 Schema 演化，加字段删字段都很方便。
- Protobuf：序列化后体积更小，解析速度更快。但它的 Schema 是 .proto 文件，需要编译成代码，集成成本稍高。
来看一个实际对比：
| 特性 | Avro | Protobuf |
|---|---|---|
| Schema 存储 | 可嵌入数据文件 | 需外部 .proto 文件 |
| Schema 演化 | 原生支持，友好 | 支持，但需手动维护 |
| 序列化大小 | 中等 | 更小 |
| 解析速度 | 中等 | 更快 |
| 与 Kafka 集成 | Confluent 原生支持 | 需额外配置 |
下面是一个 Avro Schema 的契约定义示例：
{
"type": "record",
"name": "OrderCreated",
"namespace": "com.example.order",
"fields": [
{"name": "order_id", "type": "string"},
{"name": "user_id", "type": "string"},
{"name": "amount", "type": "double"},
{"name": "items", "type": {"type": "array", "items": "string"}},
{"name": "created_at", "type": "long", "logicalType": "timestamp-millis"}
]
}
对应的 Protobuf 定义：
syntax = "proto3";
package com.example.order;
message OrderCreated {
string order_id = 1;
string user_id = 2;
double amount = 3;
repeated string items = 4;
int64 created_at = 5; // unix timestamp millis
}
你想想看，如果生产端用 Avro 写了个 amount 是 double，消费端却按 int 去读，不出错才怪。有了 Schema Registry，这种问题在注册阶段就会被拦截。
4.4 整体流程：从生产到消费的契约链路
为了让你更直观地理解，我画了一张图。它展示了从生产者注册 Schema，到消费者拉取 Schema 并反序列化的完整链路。
你看，整个链路其实不复杂。但每个环节的契约一旦缺失，就会出问题。我见过最典型的场景是：生产者注册了 Schema V1，然后偷偷改了字段类型，没注册 V2，直接往 Topic 里写。消费者拉到的还是 V1，反序列化直接抛异常。嗯，这种问题，靠人盯是盯不住的，必须靠 Schema Registry 的兼容性检查来卡住。
4.5 避坑指南
- 不要用 JSON 做实时流契约：JSON 没有强类型 Schema，字段类型、枚举值、嵌套结构全靠约定。一旦有人手抖，消费端就崩。
- Schema Registry 必须高可用：它挂了，生产者和消费者都注册不了 Schema，整个流就断了。我建议至少部署 3 节点。
- 兼容性策略要提前定：不要在线上环境来回切换。我曾经见过一个团队，从 BACKWARD 切到 FORWARD，结果老 Schema 全被拒绝，线上断流 2 小时。
- Topic 命名里带版本号：比如
order.created.v1。这样即使 Schema 不兼容，也能通过新 Topic 做灰度迁移。