Schema Registry 不是什么都能管
Confluent Schema Registry 在 Kafka 生态里名气很大，但它能管的和不能管的，边界要画清楚。Registry 的核心能力只有两条：
- 存储 schema：给每个主题-版本组合一个全局 ID，生产者发数据时只发 ID 不发完整 schema；
- 兼容性校验：在注册新 schema 时拒绝不满足预设兼容策略的变更。
它能管的是「同一主题内的 schema 演进」，不能管的是：
- 跨主题的一致性：
library.loan.created和library.loan.returned共用loanId字段，registry 不会检查两者定义是否一致； - 字段语义的漂移：
status: string的值集变了，registry 看到的是「类型没变」，自动放行； - 消费者侧的契约：registry 只校验「能不能序列化」，不校验「消费者需不需要这个字段」。
更现实的问题是：很多团队并没有部署 Schema Registry。用 docker 起一个 registry 容易，让它参与 CI、与目录对齐、在组织内推广到所有业务线，工程成本不低。本机环境也没有部署 Confluent Schema Registry——下文所有代码都不依赖它，只用 kafka-clients + Jackson + JDK 标准库。
两种无 Registry 的替代方案
没有 registry，schema 治理要靠别的机制补位。两条路径可以并存：
| 方案 | 做法 | 适用场景 | 代价 |
|---|---|---|---|
| 消息体内嵌版本信封 | 每条 Kafka value 带 schemaVersion，消费者按版本路由 | 版本数少、事件体小 | 每条消息多十几个字节 |
| 代码内版本路由表 | 包里一个 Map<Version, Parser>，版本与解析器一一对应 | 版本数可控、团队规模小 | 每增一个版本要发一次消费者 |
两种方案的共同点是「版本信息在代码和消息里，不在外部服务里」。这意味着升级版本不需要等 registry 审批，但也意味着版本不对齐时没有任何外部系统拦你——只能靠 CI 和契约测试（6.3）兜底。
随消息带 schema 版本
最小信封结构在 6.1 已经定义过：
{
"schemaVersion": "1.1.0",
"eventType": "LoanCreated",
"payload": { ... }
}
schemaVersion 用事件目录里的契约版本号，eventType 与目录的 name 一致。消费者收到消息后，先解 envelope，再按版本号选解析器。如果版本未知，按策略丢弃或进死信主题。
下面这段代码用 Jackson 实现了双版本并存——v1.0.0 的旧消息和 v1.1.0 的新消息放在同一个 topic 里，消费者统一解析：
// WithSchemaParser.java — 消息体内携带 Schema 标识的多版本解析器
import java.util.*;
import com.fasterxml.jackson.databind.*;
import com.fasterxml.jackson.databind.node.*;
public class WithSchemaParser {
static final ObjectMapper MAPPER = new ObjectMapper();
static final String V1 = "1.0.0";
static final String V1_1 = "1.1.0";
record LoanV1(
String loanId, String bookId, String memberId,
String borrowedAt, String dueAt, String occurredAt) {}
record LoanV1_1(
String loanId, String bookId, String memberId,
String borrowedAt, String dueAt, String occurredAt,
String reason) {}
record Envelope(String version, String eventType, String payload) {}
static String wrap(String version, String eventType, String payload) throws Exception {
ObjectNode node = MAPPER.createObjectNode();
node.put("schemaVersion", version);
node.put("eventType", eventType);
node.set("payload", MAPPER.readTree(payload));
return MAPPER.writeValueAsString(node);
}
static Envelope unwrap(String json) throws Exception {
JsonNode root = MAPPER.readTree(json);
String v = root.get("schemaVersion").asText();
String t = root.get("eventType").asText();
String p = MAPPER.writeValueAsString(root.get("payload"));
return new Envelope(v, t, p);
}
static Object parse(String kafkaValue) throws Exception {
Envelope env = unwrap(kafkaValue);
return switch (env.version()) {
case V1 -> MAPPER.readValue(env.payload(), LoanV1.class);
case V1_1 -> MAPPER.readValue(env.payload(), LoanV1_1.class);
default -> throw new IllegalStateException("Unknown version: " + env.version());
};
}
// 兼容读取：把任何版本统一成 v1.1.0 视图，旧消息用默认值回填
static LoanV1_1 readAsV1_1(String kafkaValue) throws Exception {
Envelope env = unwrap(kafkaValue);
return switch (env.version()) {
case V1 -> {
LoanV1 v1 = MAPPER.readValue(env.payload(), LoanV1.class);
yield new LoanV1_1(v1.loanId(), v1.bookId(), v1.memberId(),
v1.borrowedAt(), v1.dueAt(), v1.occurredAt(), "LEGACY");
}
case V1_1 -> MAPPER.readValue(env.payload(), LoanV1_1.class);
default -> throw new IllegalStateException("Unknown version: " + env.version());
};
}
public static void main(String[] args) throws Exception {
LoanV1 oldLoan = new LoanV1("L-001", "B-101", "M-1001",
"2026-10-20T09:00:00Z", "2026-11-20T09:00:00Z", "2026-10-20T09:00:00Z");
String v1env = wrap(V1, "LoanCreated",
MAPPER.writeValueAsString(oldLoan));
LoanV1_1 newLoan = new LoanV1_1("L-002", "B-102", "M-1002",
"2026-10-21T10:00:00Z", "2026-11-21T10:00:00Z", "2026-10-21T10:00:00Z",
"WALK_IN");
String v1_1env = wrap(V1_1, "LoanCreated",
MAPPER.writeValueAsString(newLoan));
System.out.println("v1 msg: " + v1env);
System.out.println("v1.1 msg: " + v1_1env);
LoanV1_1 view1 = readAsV1_1(v1env);
LoanV1_1 view2 = readAsV1_1(v1_1env);
System.out.println("v1 as v1.1: reason=" + view1.reason());
System.out.println("v1.1 as v1.1: reason=" + view2.reason());
}
}
本机实测（Temurin 25.0.5+8 / Jackson 2.22.4 / 2026-10-29）：
javac --release 21 -cp jackson-databind-2.22.4.jar:jackson-core-2.22.4.jar:jackson-annotations-2.22.jar WithSchemaParser.javajava -cp .:jackson-databind-2.22.4.jar:jackson-core-2.22.4.jar:jackson-annotations-2.22.jar WithSchemaParser
v1 msg: {"schemaVersion":"1.0.0","eventType":"LoanCreated","payload":{"loanId":"L-001","bookId":"B-101","memberId":"M-1001","borrowedAt":"2026-10-20T09:00:00Z","dueAt":"2026-11-20T09:00:00Z","occurredAt":"2026-10-20T09:00:00Z"}}
v1.1 msg: {"schemaVersion":"1.1.0","eventType":"LoanCreated","payload":{"loanId":"L-002","bookId":"B-102","memberId":"M-1002","borrowedAt":"2026-10-21T10:00:00Z","dueAt":"2026-11-21T10:00:00Z","occurredAt":"2026-10-21T10:00:00Z","reason":"WALK_IN"}}
v1 as v1.1: reason=LEGACY
v1.1 as v1.1: reason=WALK_IN
关键点：readAsV1_1 是一个兼容读取层。v1 消息没有 reason 字段，统一视图用 "LEGACY" 回填；v1.1 消息用原生值 "WALK_IN"。消费侧的投影代码只操作 LoanV1_1，不感知底层消息版本——版本差异在入口就被抹平了。
这个模式有个前提：默认值要有业务语义，不能是魔法值。"LEGACY" 表示「这条消息来自旧版本，reason 未知」，在投影表里可以正常参与排序和过滤；如果填 "" 或 null，查询时就要单独处理空值分支。
代码内版本路由表
信封方案适合「版本数少、每条消息独立解析」的场景。如果版本累积到十几个，switch 会很长；或者消费者要做「读 v1.5 数据但按 v2.0 规则折叠」，switch 就不够用了。这时候用一个显式路由表：
interface EventParser<T> {
T parse(String payload);
}
static final Map<String, EventParser<?>> PARSERS = Map.of(
"1.0.0", payload -> MAPPER.readValue(payload, LoanV1.class),
"1.1.0", payload -> MAPPER.readValue(payload, LoanV1_1.class)
);
static Object parse(String version, String payload) {
var parser = PARSERS.get(version);
if (parser == null) throw new IllegalStateException("No parser for " + version);
return parser.parse(payload);
}
路由表比 switch 更易扩展，也更容易在启动时做静态校验：遍历目录里的所有 version，检查 PARSERS 里是否都有对应条目。缺一个就报错，而不是等到消费者读到未知版本时才抛异常。
但路由表不能解决「旧消费者读新版本」的问题。如果生产端先升级到 v1.1.0，消费端的路由表还停留在 1.0.0，poll 到新消息就会抛异常。这就是升级顺序的重要性：加可选字段时，先升级消费者、再升级生产者，确保路由表在新消息到来之前已经在线。
路由表和 switch 的取舍可以总结为：
| 场景 | 推荐方案 | 原因 |
|---|---|---|
| 1–3 个版本 | switch | 简单、无额外抽象、JIT 友好 |
| 3–8 个版本 | 路由表 | 启动时静态校验、版本与解析器显式映射 |
| 8 个以上版本 | 路由表 + 解析器分包 | 减少单个类的加载体积，按需加载解析器 |
版本数超过十个通常是信号，说明主题承担的角色太多、事件粒度太粗，应该考虑按聚合或按生命周期拆分主题，而不是让解析器无限膨胀。
默认值的设计原则
兼容读取的核心是默认值。几个常见错误：
| 错误 | 后果 |
|---|---|
默认 null | 投影表出现 NULL，列表页排序和过滤都要加 IS NOT NULL |
默认 "" | 与业务上的「空字符串」语义冲突，无法区分「未知」与「确实为空」 |
默认 -1 / 0 | 数值字段的魔法值会在统计和报表里污染结果 |
| 默认当前时间 | 旧事件的默认值变成了「解析时刻」，完全丢失原意 |
好的默认值要满足两条：在业务上有明确含义，且不会与真实值混淆。reason = "LEGACY"、tier = "UNKNOWN"、delta = 0（增量为零确实有含义）都是可接受的；null、-1、"" 要尽量避免。
版本窗口与清理
消费者不可能永远维护所有历史版本的解析器。通常做法是保留一个「major 版本窗口」：
- 当前 major 版本的所有 minor 版本都要能解析；
- 上一个 major 版本的解析器保留一个观察期（如 30 天或两个发布周期）；
- 更老的版本由归档任务把事件格式迁移到当前版本，消费者不再直接消费它们。
窗口的大小取决于主题的保留期和生产者的升级节奏。如果保留期 7 天、每周发一版，窗口里最多压两三个版本；如果保留期 90 天、每月发一版，窗口可能压到十几个。窗口太大，消费者的解析器会成为技术债务；窗口太小，重建或重放历史事件时会失败。
升级顺序的策略
信封方案里，生产者和消费者的升级顺序直接影响线上稳定性。加可选字段时最安全的顺序是消费者先发版：
- 消费端升级：路由表里加入 v1.1.0 解析器，
readAsV1_1就绪； - 生产端升级：开始发送带
reason的 v1.1.0 消息； - 观察窗口：确认没有消费者报 “Unknown version” 后，旧版本解析器进入 deprecate 流程。
如果反过来，生产者先升级，而某个消费者还没上线 v1.1.0 解析器，它就会按 6.1 的兼容性判定抛异常——加可选字段虽然是 FULL 兼容，但「兼容」指的是「字节能解析」，不是「路由表里有条目」。
删字段或改类型的 major 变更不能走这个顺序，因为无论谁先升级，总有一方读不了对方的数据。major 变更必须走新主题或双写过渡（6.1）。
与事件目录的关系
无 Registry 的方案不是「不要治理」，而是把治理从运行时挪到编译时和发布时。事件目录（2.3）仍然是唯一的真相源：
- 目录里的
version字段是消费者路由表的输入； - 目录里的
compat字段是兼容性判定的依据； - 目录里的
consumers字段告诉你升级前要通知谁。
目录 + CI 检查 + 契约测试（6.3）三者组合，才能在没有 Registry 的情况下拦住破坏性变更。具体到代码层面，一个常见的 CI 门禁是「目录声明的版本必须在消费者路由表里有对应解析器」，遍历 events.yaml 的 version 字段，与 PARSERS.keySet() 做交集检查，缺一项就 fail build。
小结
- Schema Registry 能管序列化兼容，管不了语义漂移和跨主题一致性；没有 Registry 的团队要靠代码内治理补位。
- 消息体内嵌版本信封 + 兼容读取层，是无 Registry 时最轻量的方案：本机实测 v1/v1.1 双版本并存，旧消息默认值回填为
"LEGACY"。 - 代码内版本路由表适合版本数较多的场景，启动时静态校验缺漏的解析器。
- 默认值必须有业务语义，避免
null、-1、""等魔法值。 - 消费者维护一个 major 版本窗口，窗口外的旧事件由归档任务迁移格式。
- 无 Registry 不代表无治理：事件目录是真相源，CI 检查 + 契约测试是护栏。
- 消息体内嵌版本信封是最轻量方案，生产环境推荐 Kafka header 方式以降低 payload 污染。
- Schema Registry 只在多团队共享 schema 时才有显著价值，单团队可先用代码内版本表。
- 版本号变更应通过 CI 自动化验证，避免人工遗漏兼容性检查导致线上故障。
- 消费者维护一个 major 版本窗口，窗口外的旧事件由归档任务迁移格式到当前版本。
- 事件目录的
events.yaml应与代码中的解析器版本通过 CI 自动化比对校验。 - CI 门禁应检查「目录声明的版本必须在消费者路由表中有对应解析器」。
- 版本路由表的静态校验应在应用启动时执行，发现缺失解析器立即报错退出。
阅读导航：上一节 6.1 事件版本化 ｜ 下一节 6.3 消费者驱动的契约
- 版本化管理是事件驱动架构长期演进的基础保障。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。