4、Schema Registry实战：Confluent Schema Registry搭建、兼容性策略配置

好，咱们进入实战环节。前面讲了那么多理论，说白了，Schema Registry 就是给数据契约上了一把「锁」——谁想改 schema，得先过我这关。今天我就带你亲手搭一套 Confluent Schema Registry，再把兼容性策略玩明白。

核心要点：Schema Registry 不是 Kafka 的附属品，它是数据治理的「守门员」。没有它，生产者和消费者之间的 schema 早晚会打架。

4.1 环境搭建：从零跑起一个 Schema Registry

我个人习惯用 Docker Compose 快速拉起整套环境。你想想看，要是每搭一次都得手动装依赖、配路径，那得多浪费时间？

version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  schema-registry:
    image: confluentinc/cp-schema-registry:7.5.0
    depends_on:
      - kafka
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:9092
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

启动之后，验证一下是否跑通：

curl http://localhost:8081/subjects
# 应该返回空数组 []

嗯，这里要注意：Schema Registry 默认把 schema 存在 Kafka 的 _schemas 主题里。所以 Kafka 必须先启动，Registry 才能正常工作。我在项目中遇到过有人把顺序搞反了，结果 Registry 一直报连接超时。

4.2 注册第一个 Schema

咱们用 Avro 格式来演示。先定义一个用户信息 schema：

{
  "type": "record",
  "name": "User",
  "namespace": "com.example",
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": "string"}
  ]
}

通过 REST API 注册：

curl -X POST http://localhost:8081/subjects/user-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"com.example\",\"fields\":[{\"name\":\"id\",\"type\":\"int\"},{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"email\",\"type\":\"string\"}]}"}'

返回结果里会有一个 id 字段，那就是这个 schema 的版本号。第一次注册，版本就是 1。

小技巧：subject 命名规则通常是 <topic-name>-value 或 <topic-name>-key。这样 Registry 就能自动关联到对应的 Kafka 主题。我习惯在项目初期就定好这个规范，省得后期混乱。

4.3 兼容性策略：五种模式怎么选？

兼容性策略是 Schema Registry 的灵魂。说白了，它决定了「旧的消费者能不能读懂新的数据」。下面这张图帮你理清关系：

Schema Registry 兼容性策略决策树 兼容性策略 BACKWARD (默认) FORWARD FULL BACKWARD_TRANSITIVE FORWARD_TRANSITIVE FULL_TRANSITIVE 关键区别 左侧：只检查最新版本 vs 新版本 右侧：检查所有历史版本 vs 新版本（传递性）

五种策略的核心区别，我用表格给你列清楚：

策略 检查范围 典型场景 风险等级
BACKWARD 新 schema 兼容旧 schema 消费者先升级，生产者后升级 低
FORWARD 旧 schema 兼容新 schema 生产者先升级，消费者后升级 中
FULL 双向兼容 生产者和消费者同时升级 低
BACKWARD_TRANSITIVE 新 schema 兼容所有旧版本 严格治理，不允许回退 高
FORWARD_TRANSITIVE 所有旧版本兼容新 schema 长期演进，不允许跳过 高

4.4 配置兼容性策略

你可以全局配置，也可以针对单个 subject 配置。我建议在项目初期就用全局默认值，等遇到特殊需求再单独覆盖。

全局配置（在 docker-compose 里加）：

SCHEMA_REGISTRY_KAFKASTORE_TOPIC_REPLICATION_FACTOR: 1
SCHEMA_REGISTRY_COMPATIBILITY_LEVEL: FULL_TRANSITIVE

针对单个 subject 配置：

curl -X PUT http://localhost:8081/config/user-value \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"compatibility": "BACKWARD"}'

验证配置是否生效：

curl http://localhost:8081/config/user-value
# 返回：{"compatibilityLevel":"BACKWARD"}

避坑指南：我曾经在一个金融项目中，把全局策略设成了 BACKWARD_TRANSITIVE。结果业务方想加一个可选字段，但因为之前某个历史版本不允许这种变更，死活注册不上。后来我改成 BACKWARD，问题就解决了。记住：传递性策略虽然安全，但会限制灵活性。选之前一定要想清楚你的业务场景。

4.5 实战：模拟一次 Schema 演进

咱们用刚才的 User schema 来演示。假设现在要加一个 phone 字段，策略是 BACKWARD：

{
  "type": "record",
  "name": "User",
  "namespace": "com.example",
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": "string"},
    {"name": "phone", "type": ["null", "string"], "default": null}
  ]
}

注意看，我把 phone 设成了 ["null", "string"] 并给了默认值 null。为什么？因为 BACKWARD 策略要求新 schema 能兼容旧 schema——旧消费者读到新数据时，遇到不认识字段，得能按默认值处理。如果不给默认值，注册就会失败。

注册新版本：

curl -X POST http://localhost:8081/subjects/user-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"com.example\",\"fields\":[{\"name\":\"id\",\"type\":\"int\"},{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"email\",\"type\":\"string\"},{\"name\":\"phone\",\"type\":[\"null\",\"string\"],\"default\":null}]}"}'

返回版本号 2。这时候你查一下 subject 的历史：

curl http://localhost:8081/subjects/user-value/versions
# 返回：[1, 2]

嗯，到这里你应该明白了：Schema Registry 本质上就是一个版本管理器。每次兼容性检查通过，就生成一个新版本。不通过，就返回 409 冲突。

个人经验：我建议你在 CI/CD 流水线里加一步 schema 兼容性检查。比如在 PR 合并前，自动把新 schema 注册到测试环境的 Registry 上，如果返回 409，直接阻断合并。这样能避免「代码合进去了，上线才发现 schema 不兼容」的尴尬。

4.6 常见问题与排查

  • 注册返回 409 Conflict：兼容性检查未通过。检查你的 schema 变更是否违反了当前策略。比如 BACKWARD 下删字段、改字段类型都会失败。
  • 消费者反序列化失败：生产者和消费者用的 schema 版本不一致。检查一下消费者是否指定了 specific.avro.reader=true，或者是否用了 SchemaRegistryClient 自动获取最新 schema。
  • Registry 启动报错：Kafka 没起来，或者 _schemas 主题被误删了。重启 Registry 会自动重建主题，但历史数据就丢了。

我曾经在一个生产环境里遇到过 Registry 磁盘写满的情况——因为 schema 版本太多，又没有清理策略。后来我加了定时任务，把超过 30 天的旧版本删掉，问题才解决。嗯，这个坑你们可以提前避开。

← 上一章 📖 返回目录 下一章 →