《Java 消息中间件·Kafka 内核与客户端》9.1 序列化器与 Schema 边界

序列化不在 Sender 线程里

很多人对生产者线程模型有一个模糊印象：send() 把对象丢进队列，后台线程再慢慢序列化。这个印象会让后面所有关于「可变对象」的讨论跑偏。

4.3.0 的 KafkaProducer.send() 只有两行有效代码：

public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    // intercept the record, which can be potentially modified; this method does not throw exceptions
    ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
    return doSend(interceptedRecord, callback);
}

doSend() 是同步执行的。它在调用线程里依次做：等元数据、序列化 key、序列化 value、算分区、把字节追加进 RecordAccumulator 的批次，然后返回一个 Future。真正跨网络的发送由 Sender 线程负责，但序列化在 send() 返回之前就已经结束了。

这带来三个直接后果：

1. 序列化异常从 send() 抛出，不是从 Future.get() 抛出。类型不匹配会变成 SerializationException，而不是一个失败的 Future。
2. 序列化器在业务线程上跑，它的耗时直接计入 send() 的延迟。复杂的反射式序列化会把业务线程拖慢。
3. send() 返回之后修改对象，不会再影响已经写进批次的字节。这一点和很多文章的说法相反，下面用实验说明。

内置序列化器：先看清字节

Kafka 自带一组序列化器，都在 org.apache.kafka.common.serialization 包里。它们的行为很朴素，但有几个细节值得先记住。

本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）：

[内置] String('借阅')=6 字节 e5 80 9f e9 98 85
[内置] Integer(7)=00 00 00 07
[内置] Long(7L)=00 00 00 00 00 00 00 07
[内置] UUID=30 30 30 30 30 30 30 30 2d 30 30 30 30 2d ...（共 36 字节）
[内置] ByteArray 返回同一引用? true

两条容易踩的：

• UUIDSerializer 不是 16 字节二进制，而是 36 字节文本。用它当 key 时，分区哈希的输入是这 36 个字符，不是 UUID 的 128 位值。要紧凑就得自己写 16 字节版本。
• ByteArraySerializer 直接返回传入的数组，没有拷贝。源码只有一行 return data;。但因为 doSend() 会立刻把字节追加进批次，send() 返回后再改数组并不会污染消息。

可变对象：一条被传错的建议

「不要在消息里放可变对象」这句话本身没错，但常见的解释是错的。更准确的说法有三条。

第一，send() 返回后再改，通常已经晚了

本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）：用 ByteArraySerializer 连发 5 条，每次 send() 返回后立刻把 payload[1] 改成 0xEE，再消费：

trial 0 发送后立刻把 payload[1] 改成 0xEE
...
消费 k0 -> 00 01 02 03
消费 k1 -> 01 01 02 03
消费 k2 -> 02 01 02 03
消费 k3 -> 03 01 02 03
消费 k4 -> 04 01 02 03

5 条全部保持原值。原因是 RecordAccumulator.append 在 send() 内同步把 key、value、headers 编码进批次缓冲区，之后 Sender 读到的是批次里的副本。所以「send 之后改对象导致发错」这个说法，在 4.3.0 的普通发送路径上不成立。

第二，真正的危险是并发修改与 send() 重叠

序列化器在调用线程同步读对象字段。如果另一个线程同时修改同一个对象，序列化器可能读到「loanId 是新的、borrowedAt 是旧的」这种撕裂状态。单次实验很难稳定复现，但这是数据竞争，不是概率问题。

因此规则应该改成：传给 send() 的对象在 send() 执行期间不能被其他线程修改。最省事的做法是用不可变 record，字段全部 final，集合用 List.copyOf() 包一层。

第三，消费端的复用实例会让一批记录变成一条

这是最容易在生产里造成「数据看起来对、其实全错」的 bug。写一个反序列化器，内部只保留一个可变对象，每次 deserialize() 都改它的字段再返回。业务侧习惯先把一批记录收进 List 再处理：

本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）：

[复用实例] 收集 5 条: [L-5@1700000000005, L-5@1700000000005, L-5@1700000000005, L-5@1700000000005, L-5@1700000000005]
[复用实例] 不同对象数=1
[每条新建] 收集 5 条: [L-1@1700000000001, L-2@1700000000002, L-3@1700000000003, L-4@1700000000004, L-5@1700000000005]
[每条新建] 不同对象数=5

复用实例时，5 条记录在内存里是同一个对象，List 里五个引用全指向最后一次反序列化的结果。日志打印、批量入库、聚合统计都会得到「5 条相同的最后一条」。这类错误不会抛异常，只会让数据悄悄错掉。

结论：不可变要求主要落在反序列化器的返回值上，而不是发送端「send 后不要改」。

自定义序列化器的边界

自己写序列化器时，真正要实现的只有 serialize()，但另外两个生命周期方法不能忽略。假设领域事件 LoanEvent 的字段与前面的 v2 一致：

public final class LoanEventSerializer implements Serializer<LoanEvent> {
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 只读配置；不要在这里打开需要按记录关闭的资源
    }

    @Override
    public byte[] serialize(String topic, LoanEvent data) {
        if (data == null) {
            return null; // null value 在 compact topic 里表示墓碑
        }
        return Serde.encodeV2(new Serde.LoanEventV2(
                data.loanId(), data.bookId(),
                data.borrowedAt().toEpochMilli(),
                data.dueAt() == null ? null : data.dueAt().toEpochMilli(),
                data.memberTier()));
    }

    @Override
    public void close() {
        // 释放 configure 里创建的资源
    }
}

三个约束：

• 一个实例会被多个线程共用。不要在字段里缓存「当前正在序列化的对象」，也不要用非线程安全的 SimpleDateFormat。格式化时间请用 DateTimeFormatter，它是不可变的。
• serialize() 不要做 I/O。查数据库补字段、调远程服务，都会把延迟加到业务线程的 send() 上。
• key 和 value 是两个独立的序列化器，通过 key.serializer / value.serializer 分别配置。key 决定分区，格式应该更稳定、更短。

反序列化器对应三个方法：configure()、deserialize()、close()。deserialize() 返回 null 表示这条记录的 value 是 null（墓碑），不是「解析失败」。解析失败应该抛异常，让 RecordDeserializationException 带上 offset，而不是悄悄返回一个空对象。

schema 演进的最小要求

Kafka 不在 broker 端校验消息结构。topic 里可以同时躺着 v1 和 v2 的字节，broker 只当它是 byte[]。所以 schema 兼容性完全由客户端自己负责。

要让新旧版本共存，最小的四条要求是：

1. 新增字段必须是可选的。老数据没有这个字段，新读者要能给出默认值或 null。
2. 永远不要复用字段编号。删掉的编号就让它空着，否则老读者会把新语义当成旧语义。
3. 不要改字段类型，也不要改字段语义。long 改成 int、毫秒时间戳改成秒，都会让老读者读出错误的值。
4. 读者要能跳过不认识的字段。这要求格式自带长度信息，而不是「按固定顺序读到底」。

一条自制 TLV 格式

为了把这件事做实，这里手写一个最小格式，不引入任何第三方库：

头部：magic(2 字节 'LE') + version(1 字节) + fieldCount(2 字节)
字段：tag(1 字节) + type(1 字节) + length(4 字节) + payload
type：1=STRING（UTF-8），2=LONG（8 字节大端），3=INT（4 字节大端）

v1 定义三个字段：1=loanId、2=bookId、3=borrowedAt。v2 在后面追加 4=dueAt、5=memberTier，两个都是可选的。

本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）编码结果：

v1 hex = 4c 45 01 00 03 01 01 00 00 00 06 4c 2d 31 30 30 31 02 01 00 00 00 04 42 2d 34 32 03 02 00 00 00 08 00 00 01 8b cf e5 68 00
v2 hex = 4c 45 02 00 05 01 01 00 00 00 06 4c 2d 31 30 30 32 02 01 00 00 00 04 42 2d 34 32 03 02 00 00 00 08 00 00 01 8b cf e6 ee a0 04 02 00 00 00 08 00 00 01 8b d5 0b c4 00 05 01 00 00 00 04 47 4f 4c 44
v1 bytes=41 v2 bytes=65

4c 45 是 magic，紧跟着的 01 / 02 是版本号。老读者看到 version=2，可以选择拒绝，也可以选择「只读认识的 tag 1/2/3，其余跳过」。

兼容与不兼容的四种组合

同一个 topic 里写 3 条 v1、3 条 v2，再用两种老读者去读：

本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）：

[1] 严格老读者读 v1: LoanEventV1[loanId=L-1001, bookId=B-42, borrowedAt=1700000000000]
[2] 严格老读者读 v2 失败: UnsupportedOperationException: unsupported schema version 2，本消费者只认识 version 1
[3] 宽容老读者读 v2: LoanEventV1[loanId=L-1002, bookId=B-42, borrowedAt=1700000100000]（dueAt 被丢弃）
[4] 新读者读 v1: LoanEventV2[loanId=L-1001, bookId=B-42, borrowedAt=1700000000000, dueAt=null, memberTier=null]
[5] 新读者读 v2: LoanEventV2[loanId=L-1002, bookId=B-42, borrowedAt=1700000100000, dueAt=1700086400000, memberTier=GOLD]

注意「宽容」不是免费的：它读得下去，但 dueAt 被静默丢弃。如果业务依赖这个字段，兼容就只是「不崩溃」，不是「正确」。

不兼容示范：复用编号与改类型

第一个反例：v2 删掉了 borrowedAt，把 tag 3 让给 dueAt。宽容老读者不会报错，只会把到期时间当成借阅时间：

[6] 复用 tag3 被老读者误读: LoanEventV1[loanId=L-1003, bookId=B-42, borrowedAt=1700086400000]
    → 比真实 borrowedAt 1700000200000 晚了 86,200,000 ms

第二个反例：把 tag 2 的 bookId 从 STRING 改成 LONG。老读者按类型检查，直接失败：

[7] 改字段类型后老读者失败: IllegalStateException: field 2 missing or not STRING: 42

这两个反例的区别很重要：改类型会快速失败，复用编号会静默错值。后者更危险，因为它能通过所有「跑通了」的验收。

Kafka 层面的后果：毒丸记录卡住分区

格式兼容只是第一层。真正让线上停摆的，是一条无法反序列化的记录进了 topic。

上面的严格老读者用 value.deserializer 接到消费者上，消费同一批数据：

[consume StrictV1Deserializer] offset=0 key=B-42 value=LoanEventV1[loanId=L-101, ...]
[consume StrictV1Deserializer] offset=1 key=B-42 value=LoanEventV1[loanId=L-102, ...]
[consume StrictV1Deserializer] offset=2 key=B-42 value=LoanEventV1[loanId=L-103, ...]
[consume StrictV1Deserializer] 失败: org.apache.kafka.common.errors.RecordDeserializationException:
    Error deserializing VALUE for partition b4.schema-lab-0 at offset 3.
    If needed, please seek past the record to continue consumption.

3 条 v1 正常解出，第 4 条（offset 3，第一条 v2）抛 RecordDeserializationException。换成宽容读者，同一批数据 decoded=6/6。

RecordDeserializationException 提供 topicPartition()、offset()、origin()（KEY 还是 VALUE）和 headers()。普通 KafkaConsumer 不会自动跳过坏记录，position 也不前进，下一次 poll() 还会撞上同一条。应用要自己决定：报警并停止，或者 seek(tp, offset + 1) 跳过。跳过意味着丢一条，这个决定不能藏在框架里。

一个务实的组合是：

• 生产端：所有消息带版本号；新增字段可选；序列化器只做纯函数转换。
• 消费端：默认严格校验版本号；遇到不认识的版本，记录 topicPartition + offset + headers 后停止，不要静默跳过。
• 发布顺序：先升级消费者（能读新旧两版），再升级生产者（开始写新版）。反过来就会出现「新数据被老消费者拒收」。

libraryd 的约定

把上面的规则落到本书的项目里，事件格式采用三条约定：

1. 每个事件第一个字节是 schema 版本，当前是 1；
2. 字段一律 TLV，tag 一旦分配就不再复用，删除的 tag 永久保留；
3. 新增字段必须有默认值：dueAt 缺失按 borrowedAt + 30 天计算，memberTier 缺失按 STANDARD。

Loan 事件里的时间是 Instant，序列化时统一写 epoch millis；金额用 long 分。这两条都是「不要改语义」的具体化：一旦线上有数据，毫秒和秒、分和元就不能再混。

检查清单

上线前对着这张表逐条过：

本节判据

读完这一节，你应该能回答三个问题：

1. 序列化异常从哪里抛？ 从调用 send() 的线程抛出，因为 doSend() 同步完成序列化和批次追加。
2. send() 返回后修改对象会污染消息吗？ 普通路径不会，字节已经拷进批次；危险在于并发修改与 send() 重叠，以及反序列化端复用实例。
3. schema 演进哪条规则最容易被忽略、后果最隐蔽？ 复用字段编号。它不抛异常，只让老读者把新字段当成旧字段读出错误的值。

下一节换一个扩展点：拦截器。序列化器决定「字节长什么样」，拦截器决定「消息在进出客户端时被谁看过、改过、记过账」。

阅读导航：上一节 8.3 ConsumerCoordinator 与位移提交 ｜ 下一节 9.2 拦截器链与埋点