Java操作Kafka：消息序列化，Avro、JSON 格式处理​

一、引言

在现代软件开发中，消息队列是一种非常重要的组件，它可以帮助我们实现异步通信、解耦系统组件等功能。Kafka 作为一个高性能、高吞吐量的分布式消息队列系统，被广泛应用于大数据、实时数据处理等领域。在使用 Kafka 时，我们经常需要处理消息的序列化问题，而 Avro 和 JSON 是两种常见的消息序列化格式。本文将详细介绍 Java 如何操作 Kafka 进行消息序列化，以及如何处理 Avro 和 JSON 格式的消息。

二、Kafka 简介

Kafka 是一个开源的分布式流处理平台，由 LinkedIn 开发并开源。它主要用于处理高吞吐量的实时数据流，具有以下特点：

• 高吞吐量：Kafka 能够处理大量的消息，每秒可以处理数百万条消息。
• 分布式：Kafka 采用分布式架构，可以在多个节点上进行数据存储和处理，提高系统的可靠性和扩展性。
• 持久化：Kafka 将消息持久化到磁盘上，保证消息不会丢失。
• 多生产者和多消费者：Kafka 支持多个生产者和多个消费者同时工作，提高系统的并发处理能力。

三、消息序列化概述

在 Kafka 中，消息是以字节数组的形式进行传输的。因此，在将消息发送到 Kafka 之前，需要将消息对象序列化为字节数组；在从 Kafka 接收消息之后，需要将字节数组反序列化为消息对象。消息序列化的目的是将复杂的对象转换为字节数组，以便在网络中传输或存储。常见的消息序列化格式有 Avro、JSON、Protobuf 等。本文将重点介绍 Avro 和 JSON 格式的处理。

四、Java 操作 Kafka 基础

4.1 添加依赖

首先，我们需要在项目中添加 Kafka 的 Java 客户端依赖。如果你使用的是 Maven 项目，可以在 pom.xml 中添加以下依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.3.1</version>
</dependency>

4.2 生产者示例

以下是一个简单的 Kafka 生产者示例，使用 String 类型的消息：

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        // 配置 Kafka 生产者的属性
        Properties props = new Properties();
        // 指定 Kafka 集群的地址
        props.put("bootstrap.servers", "localhost:9092");
        // 指定消息的序列化器
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建 Kafka 生产者实例
        Producer<String, String> producer = new KafkaProducer<>(props);

        // 创建消息记录
        ProducerRecord<String, String> record = new ProducerRecord<>("test_topic", "key", "Hello, Kafka!");

        try {
            // 发送消息
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        System.err.println("消息发送失败: " + exception.getMessage());
                    } else {
                        System.out.println("消息发送成功，分区: " + metadata.partition() + ", 偏移量: " + metadata.offset());
                    }
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者
            producer.close();
        }
    }
}

4.3 消费者示例

以下是一个简单的 Kafka 消费者示例，消费 String 类型的消息：

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        // 配置 Kafka 消费者的属性
        Properties props = new Properties();
        // 指定 Kafka 集群的地址
        props.put("bootstrap.servers", "localhost:9092");
        // 指定消费者组的 ID
        props.put("group.id", "test_group");
        // 指定消息的反序列化器
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 创建 Kafka 消费者实例
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Collections.singletonList("test_topic"));

        try {
            while (true) {
                // 拉取消息
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("收到消息: 分区 = %d, 偏移量 = %d, 键 = %s, 值 = %s%n",
                            record.partition(), record.offset(), record.key(), record.value());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭消费者
            consumer.close();
        }
    }
}

五、Avro 格式处理

5.1 Avro 简介

Avro 是一个数据序列化系统，它具有以下特点：

• 紧凑高效：Avro 序列化后的数据体积较小，传输和存储效率高。
• 模式匹配：Avro 使用模式（Schema）来定义数据结构，在序列化和反序列化时会进行模式匹配，保证数据的一致性。
• 支持多种编程语言：Avro 支持多种编程语言，包括 Java、Python、C++ 等。

5.2 添加 Avro 依赖

如果你使用的是 Maven 项目，可以在 pom.xml 中添加以下依赖：

<dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro</artifactId>
    <version>1.11.1</version>
</dependency>

5.3 定义 Avro 模式

首先，我们需要定义一个 Avro 模式，用于描述消息的数据结构。以下是一个简单的 Avro 模式示例：

{
    "type": "record",
    "name": "User",
    "fields": [
        {
            "name": "name",
            "type": "string"
        },
        {
            "name": "age",
            "type": "int"
        }
    ]
}

将上述模式保存为 user.avsc 文件。

5.4 生成 Avro Java 类

使用 Avro 提供的工具可以根据 Avro 模式生成 Java 类。在命令行中执行以下命令：

java -jar avro-tools-1.11.1.jar compile schema user.avsc .

执行上述命令后，会在当前目录下生成一个 User.java 文件。

5.5 Avro 生产者示例

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.*;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.kafka.clients.producer.*;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Properties;

public class AvroProducerExample {
    public static void main(String[] args) throws IOException {
        // 配置 Kafka 生产者的属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

        // 创建 Kafka 生产者实例
        Producer<String, byte[]> producer = new KafkaProducer<>(props);

        // 加载 Avro 模式
        Schema.Parser parser = new Schema.Parser();
        Schema schema = parser.parse(AvroProducerExample.class.getResourceAsStream("/user.avsc"));

        // 创建 Avro 记录
        GenericRecord user = new GenericData.Record(schema);
        user.put("name", "John");
        user.put("age", 30);

        // 序列化 Avro 记录
        DatumWriter<GenericRecord> datumWriter = new SpecificDatumWriter<>(schema);
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(outputStream, null);
        datumWriter.write(user, encoder);
        encoder.flush();
        byte[] messageBytes = outputStream.toByteArray();

        // 创建消息记录
        ProducerRecord<String, byte[]> record = new ProducerRecord<>("avro_topic", "key", messageBytes);

        try {
            // 发送消息
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        System.err.println("消息发送失败: " + exception.getMessage());
                    } else {
                        System.out.println("消息发送成功，分区: " + metadata.partition() + ", 偏移量: " + metadata.offset());
                    }
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者
            producer.close();
        }
    }
}

5.6 Avro 消费者示例

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.*;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.kafka.clients.consumer.*;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class AvroConsumerExample {
    public static void main(String[] args) throws IOException {
        // 配置 Kafka 消费者的属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "avro_group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");

        // 创建 Kafka 消费者实例
        KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Collections.singletonList("avro_topic"));

        // 加载 Avro 模式
        Schema.Parser parser = new Schema.Parser();
        Schema schema = parser.parse(AvroConsumerExample.class.getResourceAsStream("/user.avsc"));

        try {
            while (true) {
                // 拉取消息
                ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, byte[]> record : records) {
                    // 反序列化 Avro 记录
                    DatumReader<GenericRecord> datumReader = new SpecificDatumReader<>(schema);
                    BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(record.value(), null);
                    GenericRecord user = datumReader.read(null, decoder);

                    System.out.printf("收到消息: 姓名 = %s, 年龄 = %d%n",
                            user.get("name").toString(), (int) user.get("age"));
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭消费者
            consumer.close();
        }
    }
}

六、JSON 格式处理

6.1 JSON 简介

JSON（JavaScript Object Notation）是一种轻量级的数据交换格式，具有以下特点：

• 易读性好：JSON 数据结构清晰，易于人类阅读和编写。
• 跨语言支持：几乎所有的编程语言都支持 JSON 数据的处理。
• 灵活性高：JSON 可以表示各种复杂的数据结构。

6.2 添加 JSON 依赖

这里我们使用 Jackson 库来处理 JSON 数据。如果你使用的是 Maven 项目，可以在 pom.xml 中添加以下依赖：

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.13.4</version>
</dependency>

6.3 JSON 生产者示例

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.*;

import java.util.Properties;

// 定义用户类
class User {
    private String name;
    private int age;

    public User(String name, int age) {
        this.name = name;
        this.age = age;
    }

    public String getName() {
        return name;
    }

    public int getAge() {
        return age;
    }
}

public class JsonProducerExample {
    public static void main(String[] args) throws Exception {
        // 配置 Kafka 生产者的属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建 Kafka 生产者实例
        Producer<String, String> producer = new KafkaProducer<>(props);

        // 创建用户对象
        User user = new User("Alice", 25);

        // 将用户对象转换为 JSON 字符串
        ObjectMapper objectMapper = new ObjectMapper();
        String jsonString = objectMapper.writeValueAsString(user);

        // 创建消息记录
        ProducerRecord<String, String> record = new ProducerRecord<>("json_topic", "key", jsonString);

        try {
            // 发送消息
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        System.err.println("消息发送失败: " + exception.getMessage());
                    } else {
                        System.out.println("消息发送成功，分区: " + metadata.partition() + ", 偏移量: " + metadata.offset());
                    }
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者
            producer.close();
        }
    }
}

6.4 JSON 消费者示例

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.*;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class JsonConsumerExample {
    public static void main(String[] args) throws Exception {
        // 配置 Kafka 消费者的属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "json_group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 创建 Kafka 消费者实例
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Collections.singletonList("json_topic"));

        // 创建 ObjectMapper 实例
        ObjectMapper objectMapper = new ObjectMapper();

        try {
            while (true) {
                // 拉取消息
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    // 将 JSON 字符串转换为用户对象
                    User user = objectMapper.readValue(record.value(), User.class);

                    System.out.printf("收到消息: 姓名 = %s, 年龄 = %d%n",
                            user.getName(), user.getAge());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭消费者
            consumer.close();
        }
    }
}

七、应用场景

7.1 Avro 应用场景

• 大数据处理：在大数据处理场景中，需要处理大量的数据，Avro 的紧凑高效特性可以减少数据传输和存储的成本。
• 跨语言数据交互：当不同的编程语言之间需要进行数据交互时，Avro 的模式匹配和跨语言支持可以保证数据的一致性和兼容性。

7.2 JSON 应用场景

• Web 开发：在 Web 开发中，JSON 是一种常用的数据交换格式，前端和后端之间可以方便地使用 JSON 进行数据传输。
• 日志记录：在日志记录中，JSON 格式可以方便地记录各种信息，并且易于解析和处理。