ZH
EnglishPortuguêsFrançaisDeutschTiếng ViệtEspañol简体中文한국어日本語
首页 / 架构与设计 / Kafka

Schema Registry 和 Avro/Protobuf 在 Kafka 与 Spring Boot 中的应用

Cris The Dev · 2026/10/12 · 8 分钟阅读 · 137

1. 概述

当多个服务同时读写一个 Kafka topic 时，最大的风险就是消息格式。只需一个 producer 改变 JSON 中的字段名称，后续 consumer 就可能立刻崩溃。Confluent Schema Registry 解决了这个问题。它集中存储 schema，为每条消息关联一个 schema ID，并阻止那些破坏兼容性的变更。

本文将指导你如何搭建 Schema Registry，使用 Avro 和 Protobuf 在 Spring Boot 3.x 中发送和接收消息，配置兼容性规则，以及编写不需要真实 Kafka 的测试。

2. Schema Registry 如何工作

Schema Registry 是一个独立的 REST 服务，与 Kafka 集群并行运行。它在内部将 schema 存储到 _schemas topic 中。每个 schema 都属于一个 subject。按照默认设置（TopicNameStrategy），topic orders 中 value 的 subject 名称为 orders-value。

当 producer 发送消息时，serializer 执行以下步骤：

  1. 在 Registry 上注册或查询 schema 以获取 schema ID。结果被缓存。
  2. 按照 Confluent 的 wire format 写入 payload：
| byte 0: magic byte (0) | byte 1-4: schema ID (int) | byte 5..: Avro/Protobuf 二进制数据 |

Consumer 读取前 5 个字节，获取 schema ID，将 schema 加载并缓存一次，然后解码 payload。消息不需要携带完整的 schema，因此大小远小于 JSON。

3. 搭建本地环境

使用 Docker Compose，运行 KRaft 模式下的 Kafka 和 Schema Registry：

services:
  kafka:
    image: confluentinc/cp-kafka:7.7.1
    ports: ["9092:9092"]
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:29093
      KAFKA_LISTENERS: PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,EXTERNAL://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

  schema-registry:
    image: confluentinc/cp-schema-registry:7.7.1
    depends_on: [kafka]
    ports: ["8081:8081"]
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKA_STORE_BOOTSTRAP_SERVERS: kafka:29092
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

运行 docker compose up -d 后，你可以检查 Registry：

curl -s http://localhost:8081/subjects
[]

结果是空列表，因为还没有任何 subject 被注册。

4. Spring Boot 中的 Avro

4.1. 依赖和插件

Confluent 的 serializer 不在 Maven Central 上。你需要添加 Confluent 的 repository：

<repositories>
    <repository>
        <id>confluent</id>
        <url>https://packages.confluent.io/maven/</url>
    </repository>
</repositories>

<dependencies>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-avro-serializer</artifactId>
        <version>7.7.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
        <version>1.12.0</version>
    </dependency>
</dependencies>

Plugin avro-maven-plugin 从 .avsc 文件生成 Java 类：

<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.12.0</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals><goal>schema</goal></goals>
            <configuration>
                <sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory>
                <outputDirectory>${project.build.directory}/generated-sources/avro</outputDirectory>
                <stringType>String</stringType>
            </configuration>
        </execution>
    </executions>
</plugin>

选项 stringType=String 可以使字符串字段生成为 java.lang.String 类型，而不是 CharSequence。这样，代码和测试中的比较会更简单。

4.2. 定义 schema

创建文件 src/main/avro/Order.avsc：

{
  "type": "record",
  "name": "Order",
  "namespace": "dev.nonla.order.avro",
  "fields": [
    { "name": "id", "type": "string" },
    { "name": "customerId", "type": "string" },
    { "name": "amount", "type": "double" }
  ]
}

运行 mvn generate-sources。插件将生成类 dev.nonla.order.avro.Order，包含 builder 和嵌入式 schema。

4.3. 配置 producer 和 consumer

spring:
  kafka:
    bootstrap-servers: localhost:9092
    properties:
      schema.registry.url: http://localhost:8081
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
    consumer:
      group-id: shipping-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      properties:
        specific.avro.reader: true

属性 schema.registry.url 放在 spring.kafka.properties 下，这样对 producer 和 consumer 都适用。属性 specific.avro.reader: true 非常重要。如果没有它，consumer 会收到 GenericRecord 而不是生成的 Order 类。

4.4. 发送和接收消息

@Service
public class OrderPublisher {

    private final KafkaTemplate<String, Order> kafkaTemplate;

    public OrderPublisher(KafkaTemplate<String, Order> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publish(String orderId, String customerId, double amount) {
        Order order = Order.newBuilder()
                .setId(orderId)
                .setCustomerId(customerId)
                .setAmount(amount)
                .build();
        kafkaTemplate.send("orders", order.getId(), order);
    }
}
@Component
public class ShippingListener {

    private static final Logger log = LoggerFactory.getLogger(ShippingListener.class);

    @KafkaListener(topics = "orders")
    public void onOrder(Order order) {
        log.info("准备发送订单 {} 给客户 {}，金额 {}",
                order.getId(), order.getCustomerId(), order.getAmount());
    }
}

首次发送时，KafkaAvroSerializer 会自动将 schema 注册到 subject orders-value。你可以检查：

curl -s http://localhost:8081/subjects/orders-value/versions/latest
{"subject":"orders-value","version":1,"id":1,"schema":"{\"type\":\"record\",\"name\":\"Order\",...}"}

此时，topic 上的消息仅携带 schema ID 1，不携带完整的 schema。

5. Spring Boot 中的 Protobuf

Protobuf 的步骤与 Avro 几乎相同，只是 serializer 和代码生成方式不同。添加以下依赖：

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-protobuf-serializer</artifactId>
    <version>7.7.1</version>
</dependency>

定义 src/main/proto/order_event.proto：

syntax = "proto3";

package dev.nonla.order;

option java_package = "dev.nonla.order.proto";
option java_multiple_files = true;

message OrderEvent {
  string id = 1;
  string customer_id = 2;
  double amount = 3;
}

你可以使用 Maven 的 protoc 插件（例如 io.github.ascopes:protobuf-maven-plugin）生成 Java 类。然后配置 serializer：

spring:
  kafka:
    producer:
      value-serializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer
    consumer:
      value-deserializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer
      properties:
        specific.protobuf.value.type: dev.nonla.order.proto.OrderEvent

属性 specific.protobuf.value.type 告诉 deserializer 目标类。如果缺少，你会收到 DynamicMessage。

应该选择哪种格式？

  • Avro 适合 Kafka 生态系统和数据管道，如 Kafka Connect 或 ksqlDB。Schema 采用 JSON 格式，演进规则基于字段名和默认值。
  • Protobuf 适合当你的组织已经使用 gRPC 时。你可以重用现有的 .proto 文件。演进规则基于字段序号。

6. Schema 演进和兼容性

6.1. 兼容性级别

每个 subject 有一个兼容性级别。默认级别是 BACKWARD。

级别 含义 安全的部署顺序
BACKWARD 新 schema 能读旧数据 Consumer 先，producer 后
FORWARD 旧 schema 能读新数据 Producer 先，consumer 后
FULL 双向兼容 任意顺序
*_TRANSITIVE 检查与所有旧版本的兼容性，不仅是上一个版本 如上

你可以通过 REST API 为每个 subject 设置级别：

curl -s -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"compatibility": "FULL_TRANSITIVE"}' \
  http://localhost:8081/config/orders-value

6.2. 一个有效的变更

假设你想在 Order 中添加 currency 字段。对于 Avro，新字段必须有默认值才能达到 FULL 兼容性：

{ "name": "currency", "type": "string", "default": "VND" }

在合并代码之前，你应该用 Registry 检查新 schema：

curl -s -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"dev.nonla.order.avro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"customerId\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"currency\",\"type\":\"string\",\"default\":\"VND\"}]}"}' \
  http://localhost:8081/compatibility/subjects/orders-value/versions/latest
{"is_compatible":true}

如果省略 "default": "VND" 部分，结果将返回 false。此时 producer 在尝试注册 schema 时会收到 409 Conflict 错误。

对于 Protobuf，对应的规则是：永远不要改变或重用字段序号。删除字段时，使用 reserved 标记以防止序号被重用：

message OrderEvent {
  reserved 2;
  reserved "customer_id";
  string id = 1;
  double amount = 3;
  string currency = 4;
}

7. 使用 mock Schema Registry 编写测试

Confluent 的 serializer 支持形如 mock://<scope> 的 URL。使用此 URL，serializer 将使用内存中的模拟 Registry。你可以测试序列化而无需 Docker：

import static org.assertj.core.api.Assertions.assertThat;

class OrderAvroSerdeTest {

    private static final Map<String, Object> CONFIG = Map.of(
            "schema.registry.url", "mock://order-serde-test",
            "specific.avro.reader", true);

    @Test
    void shouldSerializeWithWireFormatAndDeserializeBack() {
        try (var serializer = new KafkaAvroSerializer();
             var deserializer = new KafkaAvroDeserializer()) {
            serializer.configure(CONFIG, false);
            deserializer.configure(CONFIG, false);

            Order order = Order.newBuilder()
                    .setId("ORD-1001")
                    .setCustomerId("CUS-42")
                    .setAmount(250_000.0)
                    .build();

            byte[] bytes = serializer.serialize("orders", order);

            assertThat(bytes[0]).isZero();                       // magic 字节
            assertThat(ByteBuffer.wrap(bytes, 1, 4).getInt()).isPositive(); // schema ID

            Object result = deserializer.deserialize("orders", bytes);
            assertThat(result).isInstanceOf(Order.class).isEqualTo(order);
        }
    }
}

这个测试检查三件事：

  • 第一个字节是 magic 字节 0。
  • 接下来的四个字节是一个有效的 schema ID。
  • Deserializer 返回正确的 Order 类，而不是 GenericRecord。这确认了 specific.avro.reader 已生效。

如果想要完整的集成测试，你可以使用 Testcontainers 运行 Kafka 和 Schema Registry。但是，使用 mock:// 的测试运行速度很快，足以捕获大多数配置错误。

注意： 使用相同 mock:// scope 的 serializer 将共享同一个模拟 Registry。每个测试类应该使用自己的 scope 以避免相互影响。

8. 常见错误和何时不应该使用

毒药消息（Poison pill）。 不符合 wire format 的消息会导致 KafkaAvroDeserializer 在到达 listener 前抛出 SerializationException。Consumer 可能会永久卡在该 offset。你应该用 ErrorHandlingDeserializer 包装 deserializer：

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: io.confluent.kafka.serializers.KafkaAvroDeserializer

在生产环境上自动注册 schema。 默认情况下 auto.register.schemas=true，意味着任何 producer 都可以注册新 schema。在生产环境上，你应该设置 auto.register.schemas: false 和 use.latest.version: true。Schema 应该仅通过 CI/CD 流水线注册，在兼容性检查步骤之后。

意外收到 GenericRecord 或 DynamicMessage。 原因通常是忘记了 specific.avro.reader 或 specific.protobuf.value.type。后果是 listener 中立即发生 ClassCastException。

更改 Avro 中的 namespace 或 record 名称。 对于 Avro，这是一个完全不同的 schema。Specific reader 将找不到对应的类。把 namespace 视为契约中不应该改变的部分。

为了方便而设置兼容性为 NONE。 这样做会失去 Registry 的所有好处。只在实验性 topic 上使用 NONE。

注意： Schema Registry 是一个运行时依赖。Serializer 有缓存，所以 Registry 暂时故障对已知 schema 影响不大。但新启动的 producer 或缓存中没有的 schema 会立刻出错。在生产环境上应该以多实例模式运行 Registry。

什么时候不应该使用 Schema Registry：

  • 系统很小，只有一个团队同时拥有 producer 和 consumer，变化很少。此时使用 Spring 的 JSON 和 JsonSerializer 会简单得多。
  • 消息需要直接被不理解 Confluent wire format 的客户端读取，例如只读取纯 JSON 的第三方工具。
  • 你无法运行额外的基础设施服务，也没有托管的 Registry。

9. 总结

Schema Registry 将 Kafka 消息格式转变为一个有版本和自动检查的契约。你已经看到了如何：

  • 使用 Docker 搭建 Registry。
  • 在 Spring Boot 3 中使用 KafkaAvroSerializer 和 KafkaProtobufSerializer 发送和接收消息。
  • 理解包含 magic 字节和 schema ID 的 wire format。
  • 设置兼容性级别并安全地演进 schema：Avro 中新字段有默认值，Protobuf 中使用 reserved。
  • 使用 mock:// Registry 快速编写测试。

在应用到生产环境时，禁用自动 schema 注册，将兼容性检查纳入 CI，并始终用 ErrorHandlingDeserializer 包装 deserializer。这样，错误的 schema 变更会在 review 阶段立即被阻止，而不是在午夜时分破坏 consumer。

kafkaschema-registryavroprotobufspring-boot
作者的应用HoanTien.proCashback on Shopee, TikTok Shop and Lazada (Vietnam)iOS · Android · Web

同主题文章