books

《Java 消息中间件·事件驱动架构》6.1 事件版本化

事件版本号放在主题名、消息信封还是字段里，每种方案在落盘、回放与消费端的成本各不相同。本文用 Java 实测三种版本化方案的兼容性判定逻辑，覆盖加字段、删字段、改名三类变更，按 FORWARD/BACKWARD/FULL 给出可复用的判定表。

Leeting Yan 2026-10-29 7 分钟阅读 3358 字

为什么要单独讲版本化

事件契约不是一次写定就再也不变的。加字段、改语义、合并模型，这些变更在生产环境里每天都在发生。版本化的第一个问题不是「用什么语义化版本」，而是版本号这个元数据放在哪——它决定了消费者怎么知道一条消息该按哪个 schema 解析。

三种方案在工程上各有取舍：

方案	版本号位置	优点	缺点
主题版本化	`library.loan.created.v2`	消费者按 topic 订阅，天然隔离	主题爆炸，重放要跨 topic 拼接
信封版本化	Kafka value 内嵌 `{"version":"2.0.0", ...}`	一条 topic 内多版本共存	消费者必须先解 envelope 再路由
字段版本化	事件体里无显式版本，靠「字段存在性」推断	兼容旧消费者无感	无法表达语义变更，推断逻辑脆弱

主题版本化在组织变大后最痛苦。library.loan.created 有 v1/v2/v3，每个相关消费者都要改订阅列表；重放历史时还得知道事件在哪个版本时期产生。信封版本化是折中：topic 不变，版本信息在 payload 里，消费者用一个小路由器根据 version 字段选择解析器。

字段版本化常见于「我们相信不会发生破坏性变更」的系统里——消费者看到字段就解析，看不到就忽略。这个假设在团队多了、发布节奏快了以后很难维持。2.3 介绍的事件目录里，version 字段是契约级别的声明，不是运行时信封；把它同时写进消息体，才能把声明变成可执行的判据。

三种变更的兼容判定

改事件 schema 之前，先问自己：新消费者能读旧数据吗？旧消费者能读新数据吗？两条都成立才是 FULL 兼容，一条是 FORWARD 或 BACKWARD，都不成立就是破坏性变更。

本节的判定逻辑用纯 Java 实现，不依赖任何 registry。输入是两份 schema（旧、新），输出是兼容性结论与原因。下面这个 SchemaCompatChecker 覆盖了三类典型变更。

// SchemaCompatChecker.java

import java.util.*;

public class SchemaCompatChecker {

    record Field(String name, String type, boolean required, String defVal) {}
    record Schema(String version, List<Field> fields) {}

    static Schema v1() {
        return new Schema("1.0.0", Arrays.asList(
            new Field("loanId", "string", true, null),
            new Field("bookId", "string", true, null),
            new Field("memberId", "string", true, null),
            new Field("borrowedAt", "timestamp", true, null),
            new Field("dueAt", "timestamp", true, null),
            new Field("occurredAt", "timestamp", true, null)
        ));
    }

    static Schema v1_1() {
        // 加可选字段
        return new Schema("1.1.0", Arrays.asList(
            new Field("loanId", "string", true, null),
            new Field("bookId", "string", true, null),
            new Field("memberId", "string", true, null),
            new Field("borrowedAt", "timestamp", true, null),
            new Field("dueAt", "timestamp", true, null),
            new Field("occurredAt", "timestamp", true, null),
            new Field("reason", "string", false, "INITIATED")
        ));
    }

    static Schema v2() {
        // 删字段，合并 borrowedAt/dueAt 为 loanPeriod
        return new Schema("2.0.0", Arrays.asList(
            new Field("loanId", "string", true, null),
            new Field("bookId", "string", true, null),
            new Field("memberId", "string", true, null),
            new Field("loanPeriod", "string", true, null),
            new Field("occurredAt", "timestamp", true, null)
        ));
    }

    static Schema v1_2() {
        // 改名：borrowedAt -> startAt
        return new Schema("1.2.0", Arrays.asList(
            new Field("loanId", "string", true, null),
            new Field("bookId", "string", true, null),
            new Field("memberId", "string", true, null),
            new Field("startAt", "timestamp", true, null),
            new Field("dueAt", "timestamp", true, null),
            new Field("occurredAt", "timestamp", true, null)
        ));
    }

    // 旧消费者读新数据 → FORWARD
    static boolean forward(Schema oldSchema, Schema newSchema) {
        Map<String, Field> newFields = byName(newSchema.fields());
        for (Field f : oldSchema.fields()) {
            if (!newFields.containsKey(f.name())) return false;
            if (!newFields.get(f.name()).type().equals(f.type())) return false;
        }
        return true;
    }

    // 新消费者读旧数据 → BACKWARD
    static boolean backward(Schema oldSchema, Schema newSchema) {
        Map<String, Field> oldFields = byName(oldSchema.fields());
        for (Field f : newSchema.fields()) {
            if (f.required()) {
                if (!oldFields.containsKey(f.name())) return false;
                if (!oldFields.get(f.name()).type().equals(f.type())) return false;
            }
        }
        return true;
    }

    static Map<String, Field> byName(List<Field> fields) {
        Map<String, Field> m = new LinkedHashMap<>();
        for (Field f : fields) m.put(f.name(), f);
        return m;
    }

    static String reason(Schema old, Schema neu, boolean isForward) {
        Map<String, Field> oldF = byName(old.fields());
        Map<String, Field> newF = byName(neu.fields());
        if (isForward) {
            for (Field f : old.fields()) {
                if (!newF.containsKey(f.name()))
                    return "旧字段 " + f.name() + " 在新 schema 中不存在";
                if (!newF.get(f.name()).type().equals(f.type()))
                    return "旧字段 " + f.name() + " 类型不一致";
            }
            return "旧 schema 所有字段在新 schema 中保留";
        } else {
            for (Field f : neu.fields()) {
                if (f.required() && !oldF.containsKey(f.name()))
                    return "新必填字段 " + f.name() + " 在旧 schema 中不存在";
            }
            return "新 schema 必填字段在旧数据中可找到";
        }
    }

    public static void main(String[] args) {
        Schema s1 = v1(), s1_1 = v1_1(), s2 = v2(), s1_2 = v1_2();

        System.out.println("=== v1.0.0 → v1.1.0 (加可选字段) ===");
        System.out.println("FORWARD:  " + (forward(s1, s1_1) ? "COMPATIBLE" : "BREAK"));
        System.out.println("BACKWARD: " + (backward(s1, s1_1) ? "COMPATIBLE" : "BREAK"));
        System.out.println("reason:   " + reason(s1, s1_1, true));

        System.out.println("\n=== v1.0.0 → v2.0.0 (删字段) ===");
        System.out.println("FORWARD:  " + (forward(s1, s2) ? "COMPATIBLE" : "BREAK"));
        System.out.println("BACKWARD: " + (backward(s1, s2) ? "COMPATIBLE" : "BREAK"));
        System.out.println("reason:   " + reason(s1, s2, false));

        System.out.println("\n=== v1.0.0 → v1.2.0 (改名) ===");
        System.out.println("FORWARD:  " + (forward(s1, s1_2) ? "COMPATIBLE" : "BREAK"));
        System.out.println("BACKWARD: " + (backward(s1, s1_2) ? "COMPATIBLE" : "BREAK"));
        System.out.println("reason:   " + reason(s1, s1_2, true));
    }
}

本机实测（Temurin 25.0.5+8 / 2026-10-29）：javac --release 21 SchemaCompatChecker.java && java SchemaCompatChecker

=== v1.0.0 → v1.1.0 (加可选字段) ===
FORWARD:  COMPATIBLE
BACKWARD: COMPATIBLE
reason:   旧 schema 所有字段在新 schema 中保留

=== v1.0.0 → v2.0.0 (删字段) ===
FORWARD:  BREAK
BACKWARD: BREAK
reason:   旧字段 borrowedAt 在新 schema 中不存在

=== v1.0.0 → v1.2.0 (改名) ===
FORWARD:  BREAK
BACKWARD: BREAK
reason:   旧字段 borrowedAt 在新 schema 中不存在

三类变更的结果对应工程上的三种处置：

加可选字段：FULL 兼容。旧消费者看不到 reason，忽略即可；新消费者读旧数据时 reason 取默认值 "INITIATED"。
删字段：双向破坏。旧消费者读到新数据，borrowedAt 和 dueAt 会缺失；新消费者读旧数据，loanPeriod 不存在。必须走新主题或双写过渡。
改名：双向破坏。borrowedAt 变成 startAt，旧消费者找不到字段，新消费者读旧数据也找不到 startAt。这不是增加字段能解决的，改名本质等于删旧加新。

完整判定表

把常见变更按方向汇总成表，写 schema 之前先查表，比事后修生产环境便宜得多：

变更	FORWARD	BACKWARD	FULL	处置建议
加可选字段	COMPATIBLE	COMPATIBLE	COMPATIBLE	直接升级，信封标 minor
加必填字段	COMPATIBLE	BREAK	BREAK	需旧消费者同步升级，或给默认值降级为可选
删字段	BREAK	BREAK	BREAK	视为破坏性变更，走新主题或双写
改字段类型	BREAK	BREAK	BREAK	类型不一致即破坏，无中间态
改名	BREAK	BREAK	BREAK	改名 = 删 + 加，全破坏
改字段语义（类型不变）	COMPATIBLE	COMPATIBLE	COMPATIBLE	但业务破坏，需人工评审，不可自动放行

使用这张表之前，先确认生产者和消费者的升级顺序：

FULL 兼容：任何顺序都可以，先升消费者或先升生产者都没风险；
BACKWARD 兼容：必须先升级消费者，否则旧消费者遇到缺少必填字段的新数据会失败；
FORWARD 兼容：必须先升级生产者，否则新消费者读旧数据时找不到新字段；
NONE / 双向破坏：必须同步发布，或者走新主题隔离。6.2 会展开无 Registry 时的升级顺序策略。

最后一行最危险。status: string 的值从 "BORROWED"/"RETURNED" 改成 "ON_LOAN"/"CLOSED"，FORWARD/BACKWARD 在字节层面都是兼容的——字段名在、类型在——但消费者的分支判断全错了。2.3 的目录里把这类变更单列一行，要求架构评审和 ADR，就是这个原因。

版本号写进消息信封

判定逻辑跑在 CI 里，版本号本身还要写进消息体，消费者才能在运行时做出路由决策。最小信封结构：

{
  "schemaVersion": "1.1.0",
  "eventType": "LoanCreated",
  "payload": { ... }
}

schemaVersion 必须匹配事件目录里的契约版本，eventType 与目录的 name 一致。消费者收到消息后，先看 schemaVersion，再选对应的 record 解析器；如果版本未知，按策略决定是丢弃、进死信，还是回退到「字段存在性」推断。

信封可以放在两个位置：

Kafka Record Header：record.headers().add("schemaVersion", "1.1.0".getBytes())，与 payload 解耦，解析器先读 header 再决定怎么读 value；
Payload 内部：如上面的 JSON envelope，header 保持干净，但解析器必须先做一次轻量 JSON 读取才能拿到版本号。

Kafka header 在 0.11.0 引入，所有现代客户端都支持。header 方案更干净：版本号不污染业务数据结构，payload 永远是纯业务数据。本卷后续代码用 payload 内嵌方案演示，因为不依赖 Kafka header API 的可移植性更好；生产环境推荐 header 方案。

信封带来一点开销：每条消息多十几个字节。对 JSON 来说可以忽略；对 Avro/Protobuf 的紧凑格式，信封放在消息头里而不是 payload 里更重要。本卷只用 JSON 演示，规则相同。

什么时候换主题

不是每次版本升级都要换主题。换主题的成本很高：消费者改订阅、旧主题保留、重放要拼接多个 topic。只有在下面两种情况下才值得：

major 版本变更：删字段、改类型、改 key。旧消费者不可能不解码就跳过，topic 隔离是最干净的方案。
语义漂移：同一个事件名表示的事实变了。LoanCreated v1 表示「借阅创建」，v2 表示「借阅创建且已预扣库存」，消费者按名字订阅会踩坑。

加可选字段、补文档、修注释，这些都不需要换主题，信封里的 minor 版本号就够了。

版本化与位移的关系

Kafka 的位移是物理偏移，与 schema 版本无关。一个分区里可能同时存在 v1 和 v2 的消息，消费者按位移顺序读取，每条消息独立解析。这带来一个约束：消费者必须能解码它可能遇到的所有历史版本，不只是当前版本。

投影重建（3.3）是这个约束最严格的场景。重建时消费者从 topic 起点读到末尾，如果中途 schema 变过三次，解析器就要同时支持三个版本。不支持，重建就失败——不是「可能失败」，是「必然失败」。

所以版本化策略要回答的最后一个问题是：消费者最多要同时维护几个版本的解析器？ 答案越小越好，通常做法是「旧版解析器保留一个 major 版本窗口，窗口外的旧版由归档任务迁移格式」。

小结

版本号推荐写进消息信封：schemaVersion + eventType + payload。主题版本化只在 major 变更时使用。
加可选字段是 FULL 兼容的唯一常见场景；删字段、改名、改类型都是双向破坏。
改字段语义（类型不变）在兼容判定表上显示兼容，但业务上破坏，必须人工评审。
本机实测三种变更的兼容性判定：v1→v1.1 FULL 兼容，v1→v2 双向破坏，v1→v1.2 改名双向破坏。
消费者必须能解码所有历史版本，这是投影重建的硬性前提。

阅读导航：上一节 5.3 并发控制与版本｜下一节 6.2 兼容性与 Schema 管理

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页