Kafka 中的 Avro 和 Schema Registry：实际设置

Apache Avro与Schema Registry在Kafka集群中的实践集成

将Schema Registry与Kafka集群集成，可通过名为_schemas的专用主题集中管理Avro模式。在docker-compose.yaml中添加Schema Registry服务，配置集群的引导服务器，并设置Kafka-UI以可视化模式。这有助于平台从JSON消息过渡到类型化的Avro消息。

在Docker中配置Schema Registry

在docker-compose.yaml中添加schema-registry服务：

schema-registry:
  image: confluentinc/cp-schema-registry:8.1.0
  container_name: schema-registry
  hostname: schema-registry
  depends_on:
    - kafka-0
    - kafka-1
    - kafka-2
  ports:
    - "8087:8081"
  environment:
    SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka-0:9091,kafka-1:9091,kafka-2:9091
    SCHEMA_REGISTRY_HOST_NAME: schema-registry
    SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

更新kafka-ui，添加对注册表的依赖并指定其地址：

kafka-ui:
  # ... 其他设置
  depends_on:
    - schema-registry
  environment:
    KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schema-registry:8081

Schema Registry使用Kafka集群作为_schemas主题中模式的后端存储。在集群配置中，采用主从模型：写入仅通过主节点以避免竞态条件，而读取由从节点处理。

Schema Registry中的领导者选举机制

SCHEMA_REGISTRY_HOST_NAME变量定义了主节点URL的主机名，该URL被写入_schemas主题。领导者选举使用类似于Kafka消费者的group.id。

如果没有主从模型，并行请求到不同节点时：

• 客户端A向节点1发送模式，看到版本1 → 创建版本2。
• 客户端B同时向节点2发送，也看到版本1 → 创建重复的版本2。

从节点使用主题中的URL将写入请求重定向到主节点。这确保了数据一致性，无需复杂的同步机制。

在order-service中准备Avro模式

创建目录src/main/avro和文件OrderPlacedEvent.avsc：

{
  "type": "record",
  "name": "OrderPlacedEvent",
  "namespace": "io.mitochondria.order.event",
  "fields": [
    { "name": "orderId", "type": "string" },
    { "name": "email", "type": "string" },
    { "name": "productName", "type": "string" },
    { "name": "quantity", "type": "int" }
  ]
}

命名空间决定了生成类的Java包。在生成前，从DTO中移除任何现有的OrderPlacedEvent。quantity类型为int（而非与null的联合类型）符合业务逻辑。

在pom.xml中添加Confluent仓库：

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

序列化器的依赖：

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

在application.properties中：

spring.kafka.producer.value-serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
spring.kafka.producer.properties.schema.registry.url=http://localhost:8087

类生成的插件：

<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals>
                <goal>schema</goal>
            </goals>
            <configuration>
                <sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory>
                <outputDirectory>${project.basedir}/src/main/java</outputDirectory>
            </configuration>
        </execution>
    </executions>
</plugin>

运行mvn clean compile。更新kafkaTemplate中的发送方法：

kafkaTemplate.send("order-placed", orderPlacedEvent.getOrderId().toString(), orderPlacedEvent);

配置inventory-service作为生产者和消费者

将OrderPlacedEvent.avsc复制到inventory-service的src/main/avro目录。添加事件模式：

InventoryRejectedEvent.avsc：

{
  "type": "record",
  "name": "InventoryRejectedEvent",
  "namespace": "io.mitochondria.inventory.event",
  "fields": [
    { "name": "orderId", "type": "string" },
    { "name": "email", "type": "string" }
  ]
}

InventoryReservedEvent.avsc：

{
  "type": "record",
  "name": "InventoryReservedEvent",
  "namespace": "io.mitochondria.inventory.event",
  "fields": [
    { "name": "orderId", "type": "string" },
    { "name": "email", "type": "string" }
  ]
}

在application.properties中配置生产者和消费者：

spring.kafka.producer.value-serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
spring.kafka.consumer.properties.spring.deserializer.value.delegate.class=io.confluent.kafka.serializers.KafkaAvroDeserializer
spring.kafka.producer.properties.schema.registry.url=http://localhost:8087
spring.kafka.consumer.properties.schema.registry.url=http://localhost:8087
spring.kafka.consumer.properties.specific.avro.reader=true

属性specific.avro.reader=true返回生成的类而非GenericRecord。移除JSON设置和旧的事件类，然后生成新的类。

为发件箱逻辑创建DTO：

public record InventoryRejectedDto(String orderId, String email) {}
public record InventoryReservedDto(String orderId, String email) {}

在服务方法中，将Avro类映射到DTO后再写入发件箱，以避免格式混合。

关键要点

• 模式存储：Schema Registry使用Kafka主题_schemas作为后端；主从模型防止重复版本。
• 代码生成：Avro-maven-plugin基于命名空间创建强类型类，并集成到构建生命周期中。
• 序列化：KafkaAvroSerializer通过URL自动从注册表注册/检索模式。
• 消费者特定设置：specific.avro.reader=true确保类型化反序列化，无需GenericRecord。
• 竞态条件防护：写入仅通过主节点进行，从节点重定向请求。

— Editorial Team