📚 课程目录第4章 Schema Registry入门

4、Schema Registry入门：架构、API、集成与本地搭建

好，咱们进入正题。前面几章聊了Schema是什么、为什么要治理，现在该动手了。Schema Registry，说白了就是给Schema上个「中央管控系统」。我最早接触它是在做Kafka消息治理的时候，那时候团队里每人写各的Avro，生产端和消费端对不上，线上故障频发。后来上了Registry，世界清净了。

4.1 Schema Registry是什么？

Schema Registry是一个独立的服务。它负责存储、校验和分发Schema。你想想看，如果没有它，每个应用都得自己管Schema版本，改一个字段就得通知所有人。有了Registry，生产端注册Schema，消费端自动拉取，版本兼容性由服务端统一判断。

核心价值一句话：让Schema成为团队共识，而不是个人备忘录。

4.2 整体架构

我习惯把Schema Registry的架构分成三层来看：

存储层：默认用Kafka的_schemas主题持久化Schema数据。也支持内存、MySQL等后端。
服务层：RESTful API服务，处理注册、查询、兼容性检查等请求。
客户端层：集成到生产者、消费者中的库，自动与Registry交互。

下面这张图能帮你快速建立整体认知：

4.3 核心API

Registry暴露的是RESTful接口。我挑几个最常用的给你说说，这些都是日常开发高频接触的。

方法	路径	作用
`POST`	`/subjects/{subject}/versions`	注册新Schema版本
`GET`	`/subjects/{subject}/versions/latest`	获取最新Schema
`GET`	`/subjects/{subject}/versions/{version}`	获取指定版本Schema
`POST`	`/compatibility/subjects/{subject}/versions/latest`	校验Schema兼容性
`GET`	`/subjects`	列出所有Subject

小经验：Subject的命名规则很重要。我一般用<topic名称>-value或<topic名称>-key，这样一眼就能看出这个Schema属于哪个Topic。

注册一个Schema的请求示例：

POST /subjects/order-events-value/versions
Content-Type: application/vnd.schemaregistry.v1+json

{
  "schema": "{\"type\":\"record\",\"name\":\"OrderEvent\",\"namespace\":\"com.example\",\"fields\":[{\"name\":\"orderId\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"}]}"
}

返回结果会带上Schema ID和版本号：

{
  "id": 101,
  "version": 3
}

4.4 客户端集成

光有服务端不行，客户端得会用。我以Java和Python为例，给你演示最典型的集成方式。

4.4.1 Java客户端集成

在Maven里加依赖：

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>7.5.0</version>
</dependency>

生产者配置：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "http://localhost:8081");

KafkaProducer<String, OrderEvent> producer = new KafkaProducer<>(props);

消费者配置类似，只是把Serializer换成Deserializer。嗯，这里要注意：schema.registry.url这个参数必须配，否则客户端不知道去哪拉Schema。

4.4.2 Python客户端集成

Python这边用confluent-kafka库：

from confluent_kafka import Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer

schema_registry_client = SchemaRegistryClient({'url': 'http://localhost:8081'})
avro_serializer = AvroSerializer(schema_registry_client, schema_str)

producer = Producer({'bootstrap.servers': 'localhost:9092'})

我曾经踩过的坑：客户端版本和服务端版本必须匹配。有一次我把客户端升到了7.6，服务端还是7.4，结果序列化时一直报兼容性错误。查了半天才发现是版本不兼容。升级服务端后问题解决。

4.5 本地环境搭建

搭建本地环境，我推荐用Docker Compose。这是最快的方式，几分钟就能跑起来。

4.5.1 Docker Compose配置

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  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

  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

4.5.2 启动与验证

执行启动命令：

docker-compose up -d

验证服务是否正常：

curl http://localhost:8081/subjects

如果返回空数组[]，说明Registry已经跑起来了。这时候你可以注册第一个Schema试试。

我个人的习惯：本地开发时，我会把SCHEMA_REGISTRY_KAFKASTORE_TOPIC_REPLICATION_FACTOR设为1。因为本地只有一个Kafka节点，副本因子设大了会报错。生产环境记得改回3。

4.5.3 常用管理命令

日常开发中，这几个命令你肯定会用到：

查看所有Subject：curl localhost:8081/subjects
查看Subject的版本列表：curl localhost:8081/subjects/order-events-value/versions
查看Schema内容：curl localhost:8081/subjects/order-events-value/versions/latest
删除Subject：curl -X DELETE localhost:8081/subjects/order-events-value

注意：删除Subject是永久操作，不可恢复。我建议你在删除前先确认没有生产者和消费者还在使用这个Schema。否则，线上会直接报反序列化错误。

好了，到这里你已经掌握了Schema Registry的核心概念、API用法、客户端集成和本地搭建。下一步，你可以试着注册几个不同版本的Schema，看看兼容性校验是怎么工作的。动手试一下，比看十遍文档都管用。