（七）消息队列-Kafka 序列化avro（传递）

客从远方来，遗我双鲤鱼。呼儿烹鲤鱼，中有尺素书。
——佚名《饮马长城窟行》

在这里插入图片描述

本文已同步CSDN、掘金平台、知乎等多个平台，图片依然保持最初发布的水印（如CSDN水印）。（以后属于本人原创均以新建状态在多个平台分享发布）

前言

多年前，由于工作的性质，发现这系列没有写完，想了想，做人做事还是要有始有终。🤣实在是借口太多了，太不像话了…由于时间过得太久了，这篇开始，可能很多技术以最新或最近的几个版本为主了。

问题背景

在Kafka中，生产者与消费者之间传输消息时，通常需要对数据进行序列化和反序列化。常见的序列化方式如JSON或String存在以下问题：

数据冗余：字段名重复存储，占用带宽；
兼容性差：新增或删除字段时容易导致上下游解析失败；
类型安全缺失：动态解析易引发运行时错误。

而Avro作为一种高效的二进制序列化框架，通过Schema定义数据结构，可实现紧凑存储、动态兼容性和强类型校验，成为Kafka生态中推荐的序列化方案27。

核心原理

Schema驱动
Avro要求所有数据必须与预定义的Schema文件（.avsc）匹配。Schema以JSON格式描述数据结构，例如：

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

然后使用 avro-maven-plugin 生成 Java 类：

<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.11.0</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals>
                <goal>schema</goal>
            </goals>
        </execution>
    </executions>
</plugin>

执行 mvn clean compile 后，com.example.avro.User 类会被自动生成。

生产者与消费者需共享同一Schema，确保序列化与反序列化的一致性。

二进制编码
Avro将数据转换为紧凑的二进制格式，相比JSON减少约30%-50%的存储与传输开销。例如，整型字段直接以二进制存储，无需字段名冗余7。
Schema Registry
为实现Schema动态管理，通常搭配Schema Registry（如Confluent或Apicurio）使用。其核心功能包括：
- Schema版本控制与兼容性检查；
- 通过唯一ID关联消息与Schema，避免传输完整Schema带来的性能损耗。

实现步骤

以下以Java代码为例，展示Kafka集成Avro的配置方法：

1. 添加依赖

<dependencies>
    <!-- Spring Kafka 依赖-->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    
    <!-- Avro 依赖 -->
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
    </dependency>
    
    <!-- Schema Registry 依赖 -->
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-avro-serializer</artifactId>
        <version>7.2.1</version>
    </dependency>
</dependencies>

运行 HTML

2. 配置生产者

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", StringSerializer.class.getName());
props.put("value.serializer", KafkaAvroSerializer.class.getName());
props.put("schema.registry.url", "http://localhost:8081"); // Schema Registry地址

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

// 构建Avro消息
GenericRecord user = new GenericData.Record(schema);
user.put("id", 1);
user.put("name", "Alice");

producer.send(new ProducerRecord<>("user-topic", user));



------ SpringBoot框架 直接用配置application.yml 和生产者服务类--------------
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
      
@Service
public class UserProducer {
    private final KafkaTemplate<String, User> kafkaTemplate;
    
    @Value("${kafka.topic.user}")
    private String topic;

    public UserProducer(KafkaTemplate<String, User> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendUser(User user) {
        kafkaTemplate.send(topic, user.getId().toString(), user);
    }
}

在 Spring Boot 启动后，我们可以使用以下代码发送一个 User 消息：
User user = User.newBuilder()
    .setId(1)
    .setName("Alice")
    .build();
userProducer.sendUser(user);

控制台应该能够看到消费者成功接收到 User 数据

3. 配置消费者

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "avro-consumer");
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", KafkaAvroDeserializer.class.getName());
props.put("schema.registry.url", "http://localhost:8081");

Consumer<String, GenericRecord> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("user-topic"));

while (true) {
    ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, GenericRecord> record : records) {
        System.out.println("Received: " + record.value().get("name"));
    }
}

------ SpringBoot框架 直接用配置application.yml 和消费者服务类--------------
在 application.yml 中配置消费者参数：

spring:
  kafka:
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      properties:
        specific.avro.reader: true

然后编写 Kafka 消费者代码：

@Service
@KafkaListener(topics = "user_topic", groupId = "user_group")
public class UserConsumer {

    @KafkaHandler
    public void consume(User user) {
        System.out.println("Received user: " + user.getName());
    }
}

常见问题与解决方案

Schema兼容性错误
- 现象：生产者更新Schema后消费者无法解析旧数据。
- 解决：在Schema Registry中配置兼容性策略（如BACKWARD），允许新增字段并设置默认值7。
ClassNotFoundException
- 现象：反序列化时提示Avro生成的类不存在。
- 解决：通过Maven插件avro-maven-plugin自动生成Java类，并确保生成路径在编译范围内2。
性能瓶颈
- 现象：高吞吐场景下序列化延迟较高。
- 优化：复用DatumWriter和DatumReader实例，避免重复初始化开销7。