Kafka 中的消息序列化

2025年5月14日 | 11 分钟阅读

Apache Kafka 已成为现代数据管道的支柱，支持大规模的实时信息流。其结构通过解耦生产者和消费者，实现了众多结构的无缝集成。然而，要有效利用 Kafka 的能力，必须掌握消息序列化的概念——这是一个将数据转换为 Kafka 可以传输和存储的字节流的过程。

序列化确保 Kafka 能够处理各种记录类型，从简单的字符串到复杂的对象。本指南将深入探讨 Kafka 中的消息序列化，涵盖理论、实际实现、性能调优和最佳实践，以确保您基于 Kafka 的应用程序高效、可扩展且具有弹性。

1. 理解 Kafka：简要概述

在深入了解序列化之前，理解 Kafka 的架构至关重要。

生产者 (Producers)： 这些实体负责向 Kafka 主题发送数据（消息）。每条消息在发送前都会被序列化为一个字节数组。
消费者 (Consumers)： 这些实体从 Kafka 主题读取消息。接收后，消息被反序列化为原始记录的格式。
主题 (Topics)： Kafka 将数据流存储在称为主题的类别中。
代理 (Brokers)： Kafka 集群由多个 Kafka 代理组成，每个代理负责存储数据和处理客户请求。
分区 (Partitions)： Kafka 主题被划分为分区，以实现并行处理和可扩展性。

序列化是这些组件之间的桥梁，确保数据可以在生产者、代理和消费者之间平稳高效地流动。

2. 序列化在 Kafka 中的作用

在 Kafka 中，序列化在数据的传输和存储方式中扮演着重要角色。Kafka 将所有消息存储为字节数组，无论生产者使用何种数据、类型或格式。因此，发送到 Kafka 的每一条数据都必须序列化为字节数组，反之，消费者检索的每一条数据都必须反序列化回其原始形式。

序列化在 Kafka 中的关键职责

数据完整性： 确保数据在传输过程中保持完整。
兼容性： 确保生产者和消费者之间的数据格式一致，从而实现跨系统通信。
效率： 优化数据格式，以便在网络上传输和在 Kafka 代理中存储。

没有正确的序列化，数据可能会损坏、不兼容或因性能瓶颈和潜在的系统故障而处理效率低下。

3. 序列化格式的类型

Kafka 支持多种序列化格式，每种格式都适用于不同的用例和性能要求。选择正确的序列化格式至关重要，因为它会影响您基于 Kafka 的系统的效率、兼容性和可扩展性。

3.1. 字符串序列化

字符串序列化是最直接的序列化格式，将信息转换为字符串。它通常用于传输文本数据，例如日志或简单的键值对。

优点

简单性： 易于实现和理解。
人类可读： 数据保持可读格式，这对于调试很有用。

缺点

效率低下： 不如二进制格式紧凑，导致更高的存储和传输成本。
缺乏结构： 不适用于复杂统计数据。

实施

Kafka 提供了一个开箱即用的 StringSerializer

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");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "Hello, Kafka!");
producer.send(record);   

在这里，键和值都被序列化为字符串，这使其适用于数据不需要复杂结构的简单应用程序。

3.2. 字节数组序列化

字节数组序列化比字符串序列化更灵活，因为它可以同时处理字节数组，使其适用于二进制统计数据或当序列化格式未预先确定时。

优点

多功能性： 可以处理任何可以转换为字节数组的数据。
效率： 对于二进制数据，比字符串序列化更高效。

缺点

复杂性： 需要指导管理数据与字节数组之间的转换。
缺乏结构： 没有内置的模式强制执行，可能导致数据解释问题。

实施

Kafka 的 ByteArraySerializer 使用如下

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(props);
byte[] key = "key".getBytes();
byte[] value = "Hello, Kafka!".getBytes();
ProducerRecord<byte[], byte[]> record = new ProducerRecord<>("my-topic", key, value);
producer.send(record);   

字节数组序列化在数据格式为二进制或需要外部定义（例如在加密或压缩用例中）的场景中特别有用。

3.3. JSON 序列化

由于其人类可读的格式和易用性，JSON 序列化是微服务和网络应用程序中数据交换的热门选择。尽管 Kafka 本身不支持 JSON 序列化，但使用像 Jackson 这样的库很容易实现。

优点

人类可读： 易于调试和记录。
广泛支持： JSON 几乎被所有编程语言支持。

缺点

大小： 比二进制格式大，导致存储和网络开销增加。
性能： 与二进制格式相比，序列化/反序列化速度较慢。

实施

您可以使用 Jackson 实现 JSON 序列化

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Serializer;

public class JsonSerializer<T> implements Serializer<T> {
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public byte[] serialize(String topic, T data) {
        try {
            return objectMapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new SerializationException("Error serializing JSON message", e);
        }
    }
}   

这个自定义的 JsonSerializer 在将 Java 对象发送到 Kafka 之前将其转换为 JSON 格式。要使用它，您需要配置您的生产者

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "com.example.JsonSerializer");

KafkaProducer<String, MyObject> producer = new KafkaProducer<>(props);
MyObject obj = new MyObject("John", 30);
ProducerRecord<String, MyObject> record = new ProducerRecord<>("my-topic", "user1", obj);
producer.send(record);   

这种方法在 RESTful 微服务中特别有用，其中 JSON 通常用于服务之间的通信。

3.4. Avro 序列化

Avro 序列化是一种基于模式的序列化技术，由 Apache 开发。其二进制格式紧凑且快速，使其成为高吞吐量 Kafka 应用程序的理想选择。它还支持模式演进，允许在不破坏兼容性的情况下更改数据结构。

优点

紧凑： 在大小方面比 JSON 更高效。
模式强制执行： 确保数据遵循预定义的模式，防止数据损坏。
模式演进： 支持模式随时间的变化。

缺点

复杂性： 需要处理模式，这增加了开销。
依赖于模式注册中心： Avro 序列化通常需要一个模式注册中心来处理模式，增加了设置的复杂性。

实施

首先，定义您的 Avro 模式

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

接下来，使用 Avro 实现 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", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "https://:8081");

KafkaProducer<String, User> producer = new KafkaProducer<>(props);
User user = User.newBuilder().setName("John").setAge(30).setEmail("john@example.com").build();
ProducerRecord<String, User> record = new ProducerRecord<>("my-topic", "user1", user);
producer.send(record);   

对于模式管理和事实完整性至关重要的企业级应用程序，强烈推荐使用 Avro 序列化。

3.5. Protocol Buffers (Protobuf) 序列化

Protocol Buffers (Protobuf) 是由 Google 开发的另一种基于模式的序列化格式。它以其极快的速度和紧凑性而闻名，使其适用于性能关键型应用程序。

优点

效率： 非常紧凑和快速。
跨语言兼容性： Protobuf 被许多编程语言支持，使其成为跨平台系统的理想选择。
模式演进： 与 Avro 类似，Protobuf 支持模式演进。

缺点

复杂性： 需要处理模式和 Protobuf 编译器。
学习曲线： 与像 JSON 这样简单的格式相比，更难理解和实现。

实施

定义您的 Protobuf 模式 (user.Proto)

syntax = "proto3";

message User {
    string name = 1;
    int32 age = 2;
    string email = 3;
}   

使用 Protobuf 编译器从此模式生成 Java 类，然后实现 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", "com.google.protobuf.ByteString");

KafkaProducer<String, User> producer = new KafkaProducer<>(props);
User user = User.newBuilder().setName("John").setAge(30).setEmail("john@example.com").build();
ProducerRecord<String, User> record = new ProducerRecord<>("my-topic", "user1", user.toByteArray());
producer.send(record);   

在性能和跨语言兼容性至关重要的系统中，Protobuf 特别有用。

3.6. Thrift 序列化

Thrift 是由 Facebook 开发的一种序列化格式，旨在改进跨语言服务。它与 Protobuf 类似，但在定义服务和数据序列化方面提供了更大的灵活性。

优点

灵活性： 支持数据序列化和服务定义，使其成为面向服务的体系结构 (SOA) 的理想选择。
效率： 紧凑且快速，与 Protobuf 类似。
跨语言兼容性： 被多种语言支持，便于跨平台通信。

缺点

复杂性： 与较简单的序列化格式相比，设置和管理更复杂。
依赖于 Thrift 编译器： 需要 Thrift 编译器从模式生成代码，增加了开发开销。

实施

定义您的 Thrift 模式 (person.Thrift)

struct User {
  1: string name,
  2: i32 age,
  3: string email
}   

使用 Thrift 编译器生成 Java 类，然后实现 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.thrift.TSerializer");

KafkaProducer<String, User> producer = new KafkaProducer<>(props);
User user = new User().setName("John").setAge(30).setEmail("john@example.com");
TSerializer serializer = new TSerializer();
byte[] serializedUser = serializer.serialize(user);
ProducerRecord<String, byte[]> record = new ProducerRecord<>("my-topic", "user1", serializedUser);
producer.send(record);   

3.7. 自定义序列化格式

在某些情况下，内置的序列化格式可能无法满足您的特定要求。在这种情况下，实现自定义序列化格式允许您根据特定需求定制序列化过程。

实施

要实现自定义序列化器，您需要扩展 Serializer 接口

Import org.apache.kafka.common.serialization.Serializer;

public class CustomObjectSerializer implements Serializer<CustomObject> {
    @Override
    public byte[] serialize(String topic, CustomObject data) {
        // Custom serialization logic
        return data.toByteArray();
    }

    @Override
    public void close() {
        // Cleanup resources if needed
    }
}   

同样，通过扩展 Deserializer 接口来实现自定义反序列化器

import org.apache.kafka.common.serialization.Deserializer;
public class CustomObjectDeserializer implements Deserializer<CustomObject> {
    @Override
    public CustomObject deserialize(String topic, byte[] data) {
        // Custom deserialization logic
        return CustomObject.fromByteArray(data);
    }

    @Override
    public void close() {
        // Cleanup resources if needed
    }
}   

自定义序列化通常用于处理专有数据格式或实现特定统计类型的性能优化。

4. 在 Kafka 中实现序列化

在 Kafka 中实现序列化涉及配置生产者和消费者以使用适当的序列化器和反序列化器。本节将引导您完成基本设置和针对更复杂用例的高级配置。

4.1. 使用序列化设置 Kafka 生产者

要向 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");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        exception.printStackTrace();
    } else {
        System.out.println("Sent message: " + metadata);
    }
});   

在这种情况下，生产者被配置为使用 Kafka 内置的 StringSerializer 将键和值都序列化为字符串。

4.2. 使用反序列化设置 Kafka 消费者

同样，要从 Kafka 消费消息，您需要为消费者配置正确的反序列化器。

示例：基本消费者配置

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("Consumed record with key %s and value %s%n", record.key(), record.value());
    }
}   

5. Kafka 中的自定义序列化

虽然 Kafka 为常见数据类型提供了内置的序列化器和反序列化器，但在某些场景下，您需要实现自定义序列化逻辑。这在处理专有数据格式或需要优化性能或存储效率时尤其如此。

5.1. 创建自定义序列化器

要创建自定义序列化器，请实现 Kafka 提供的 Serializer 接口。以下示例展示了如何序列化一个自定义对象

import org.apache.kafka.common.serialization.Serializer;
import java.nio.ByteBuffer;

public class CustomObjectSerializer implements Serializer<CustomObject> {

    @Override
    public byte[] serialize(String topic, CustomObject data) {
        byte[] name;
        int nameSize;

        if (data == null) {
            return null;
        }

        if (data.getName() != null) {
            name = data.getName().getBytes();
            nameSize = name.length;
        } else {
            name = new byte[0];
            nameSize = 0;
        }

        ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + nameSize);
        buffer.putInt(data.getId());
        buffer.putInt(nameSize);
        buffer.put(name);

        return buffer.array();
    }
}   

在这种情况下，CustomObjectSerializer 将一个 CustomObject 转换为字节数组。序列化过程将 id 和 name 字段组合成一个可以发送到 Kafka 的单个字节缓冲区。

5.2. 创建自定义反序列化器

要反序列化自定义项，请实现 Deserializer 接口

import org.apache.kafka.common.serialization.Deserializer;
import java.nio.ByteBuffer;

public class CustomObjectDeserializer implements Deserializer<CustomObject> {

    @Override
    public CustomObject deserialize(String topic, byte[] data) {
        if (data == null || data.length == 0) {
            return null;
        }

        ByteBuffer buffer = ByteBuffer.wrap(data);
        int id = buffer.getInt();
        int nameSize = buffer.getInt();

        byte[] nameBytes = new byte[nameSize];
        buffer.get(nameBytes);

        String name = new String(nameBytes);

        return new CustomObject(id, name);
    }
}   

5.3. 处理模式演进

模式演进是序列化的一个重要方面，尤其是在数据结构随时间演变的系统中。Avro 和 Protobuf 提供了对模式演进的内置支持，允许您添加、删除或更改字段而不会破坏现有的消费者。

示例：Avro 模式演进

假设您有一个初始的 Avro 模式

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

后来，您需要添加一个新字段 email

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

6. 序列化性能考量

序列化性能可以显著影响您基于 Kafka 的系统的吞吐量和延迟。选择正确的序列化格式并优化序列化过程对于实现高性能至关重要。

性能优化技巧

使用紧凑格式： 对于高吞吐量应用程序，优先选择像 Avro 或 Protobuf 这样的二进制格式，而不是像 JSON 这样的文本格式。
优化序列化逻辑： 最小化自定义序列化逻辑的复杂性，以减少 CPU 开销。
批量发送消息： 分批发送消息以减少与单个消息处理相关的开销。

7. 序列化和反序列化中的错误处理

处理序列化和反序列化中的错误对于确保数据完整性和系统可靠性至关重要。错误可能由于各种原因发生，包括不兼容的数据格式、损坏的数据或自定义序列化器和反序列化器中的错误。

常见的序列化和反序列化错误

SerializationException： 当消息无法序列化时抛出。
DeserializationException： 当消息无法反序列化时抛出。
ClassCastException： 当反序列化的对象无法转换为预期类型时发生。

错误处理策略

优雅降级： 记录错误并继续处理其他消息，特别是在非关键应用程序中。
模式验证： 在序列化之前验证模式，以便在过程早期捕获潜在错误。
监控和警报： 设置监控和警报，以及时发现并响应序列化错误。

示例：处理反序列化错误

try {
    CustomObject obj = customDeserializer.deserialize("my-topic", record.value());
} catch (DeserializationException e) {
    System.err.println("Failed to deserialize message: " + e.getMessage());
}   

有效地处理错误可以确保您基于 Kafka 的应用程序具有弹性，并且能够克服序列化问题而不会导致数据丢失或系统停机。

8. Kafka 中消息序列化的最佳实践

遵循 Kafka 中消息序列化的最佳实践有助于您构建健壮、高效和可扩展的数据管道。

关键最佳实践

选择正确的序列化格式：根据您的用例，选择一种在清晰度、性能和兼容性之间取得平衡的序列化格式。

仔细管理模式： 使用模式注册中心来管理模式并以受控的方式处理模式演进。
监控序列化性能： 持续监控序列化性能，并根据需要进行优化以维持系统效率。
优雅地处理错误： 实施健壮的错误管理，以确保序列化问题不会中断您的数据管道。

9. 结论

Kafka 中的消息序列化是构建可扩展、高效和可靠数据管道的基础组件。通过学习各种序列化格式、实现自定义序列化器并遵循最佳实践，您可以确保基于 Kafka 的系统在性能和弹性方面得到优化。

无论您是处理简单的字符串消息还是具有演进模式的复杂对象，本指南中包含的原则和策略都将帮助您有效地管理 Kafka 应用程序中的序列化。采用正确的方法，您可以最大限度地发挥 Kafka 作为一个强大的实时数据流平台的潜力，能够满足现代数据驱动应用程序的需求。

下一主题Zookeeper-in-kafka

Kafka 中的消息序列化

1. 理解 Kafka：简要概述

2. 序列化在 Kafka 中的作用

序列化在 Kafka 中的关键职责

3. 序列化格式的类型

3.1. 字符串序列化

3.2. 字节数组序列化

3.3. JSON 序列化

3.4. Avro 序列化

3.5. Protocol Buffers (Protobuf) 序列化

3.6. Thrift 序列化

3.7. 自定义序列化格式

4. 在 Kafka 中实现序列化

4.1. 使用序列化设置 Kafka 生产者

4.2. 使用反序列化设置 Kafka 消费者

5. Kafka 中的自定义序列化

5.1. 创建自定义序列化器

5.2. 创建自定义反序列化器

5.3. 处理模式演进

6. 序列化性能考量

性能优化技巧

7. 序列化和反序列化中的错误处理

常见的序列化和反序列化错误

错误处理策略

示例：处理反序列化错误

8. Kafka 中消息序列化的最佳实践

关键最佳实践

9. 结论

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Kafka 教程

Kafka 安装

Kafka CLI

Kafka 编程

实时示例

Kafka 监控

Kafka Connect

Kafka Streams

杂项

Kafka 中的消息序列化

1. 理解 Kafka：简要概述

2. 序列化在 Kafka 中的作用

序列化在 Kafka 中的关键职责

3. 序列化格式的类型

3.1. 字符串序列化

3.2. 字节数组序列化

3.3. JSON 序列化

3.4. Avro 序列化

3.5. Protocol Buffers (Protobuf) 序列化

3.6. Thrift 序列化

3.7. 自定义序列化格式

4. 在 Kafka 中实现序列化

4.1. 使用序列化设置 Kafka 生产者

4.2. 使用反序列化设置 Kafka 消费者

5. Kafka 中的自定义序列化

5.1. 创建自定义序列化器

5.2. 创建自定义反序列化器

5.3. 处理模式演进

6. 序列化性能考量

性能优化技巧

7. 序列化和反序列化中的错误处理

常见的序列化和反序列化错误

错误处理策略

示例：处理反序列化错误

8. Kafka 中消息序列化的最佳实践

关键最佳实践

9. 结论

相关帖子

Kafka 控制器

构建从 Kafka 到 Cosmos DB 的复制管道

Kafka 复制

Kafka 记录

使用 Kafka 和 ML 进行自动化异常检测

使用 Kafka 的 BigQuery 数据仓库

Kafka Streams 的时间旅行查询

Kafka 中的系统监控和警报

Apache Kafka 消息键

AI 驱动的 Kafka 代理优化

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器