Kafka消息序列化和反序列化(下)

简介: 接上一篇:Kafka消息序列化和反序列化(上)。有序列化就会有反序列化,反序列化的操作是在Kafka Consumer中完成的,使用起来只需要配置一下key.deserializer和value.deseriaizer。

接上一篇:Kafka消息序列化和反序列化(上)

有序列化就会有反序列化,反序列化的操作是在Kafka Consumer中完成的,使用起来只需要配置一下key.deserializer和value.deseriaizer。对应上面自定义的Company类型的Deserializer就需要实现org.apache.kafka.common.serialization.Deserializer接口,这个接口同样有三个方法:

  1. public void configure(Map<String, ?> configs, boolean isKey):用来配置当前类。
  2. public byte[] serialize(String topic, T data):用来执行反序列化。如果data为null建议处理的时候直接返回null而不是抛出一个异常。
  3. public void close():用来关闭当前序列化器。

下面就来看一下DemoSerializer对应的反序列化的DemoDeserializer,详细代码如下:

public class DemoDeserializer implements Deserializer<Company> {
    public void configure(Map<String, ?> configs, boolean isKey) {}
    public Company deserialize(String topic, byte[] data) {
        if (data == null) {
            return null;
        }
        if (data.length < 8) {
            throw new SerializationException("Size of data received by DemoDeserializer is shorter than expected!");
        }
        ByteBuffer buffer = ByteBuffer.wrap(data);
        int nameLen, addressLen;
        String name, address;
        nameLen = buffer.getInt();
        byte[] nameBytes = new byte[nameLen];
        buffer.get(nameBytes);
        addressLen = buffer.getInt();
        byte[] addressBytes = new byte[addressLen];
        buffer.get(addressLen);
        try {
            name = new String(nameBytes, "UTF-8");
            address = new String(addressBytes, "UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new SerializationException("Error occur when deserializing!");
        }
        return new Company(name,address);
    }
    public void close() {}
}

有些读者可能对新版的Consumer不是很熟悉,这里顺带着举一个完整的消费示例,并以DemoDeserializer作为消息Value的反序列化器。

Properties properties = new Properties();
properties.put("bootstrap.servers", brokerList);
properties.put("group.id", consumerGroup);
properties.put("session.timeout.ms", 10000);
properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("value.deserializer", "com.hidden.client.DemoDeserializer");
properties.put("client.id", "hidden-consumer-client-id-zzh-2");
KafkaConsumer<String, Company> consumer = new KafkaConsumer<String, Company>(properties);
consumer.subscribe(Arrays.asList(topic));
try {
    while (true) {
        ConsumerRecords<String, Company> records = consumer.poll(100);
        for (ConsumerRecord<String, Company> record : records) {
            String info = String.format("topic=%s, partition=%s, offset=%d, consumer=%s, country=%s",
                    record.topic(), record.partition(), record.offset(), record.key(), record.value());
            System.out.println(info);
        }
        consumer.commitAsync(new OffsetCommitCallback() {
            public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                if (exception != null) {
                    String error = String.format("Commit failed for offsets {}", offsets, exception);
                    System.out.println(error);
                }
            }
        });
    }
} finally {
    consumer.close();
}

有些时候自定义的类型还可以和Avro、ProtoBuf等联合使用,而且这样更加的方便快捷,比如我们将前面Company的Serializer和Deserializer用Protostuff包装一下,由于篇幅限制,笔者这里只罗列出对应的serialize和deserialize方法,详细参考如下:

public byte[] serialize(String topic, Company data) {
    if (data == null) {
        return null;
    }
    Schema schema = (Schema) RuntimeSchema.getSchema(data.getClass());
    LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
    byte[] protostuff = null;
    try {
        protostuff = ProtostuffIOUtil.toByteArray(data, schema, buffer);
    } catch (Exception e) {
        throw new IllegalStateException(e.getMessage(), e);
    } finally {
        buffer.clear();
    }
    return protostuff;
}

public Company deserialize(String topic, byte[] data) {
    if (data == null) {
        return null;
    }
    Schema schema = RuntimeSchema.getSchema(Company.class);
    Company ans = new Company();
    ProtostuffIOUtil.mergeFrom(data, ans, schema);
    return ans;
}

如果Company的字段很多,我们使用Protostuff进一步封装一下的方式就显得简洁很多。不过这个不是最主要的,而最主要的是经过Protostuff包装之后,这个Serializer和Deserializer可以向前兼容(新加字段采用默认值)和向后兼容(忽略新加字段),这个特性Avro和Protobuf也都具备。

自定义的类型有一个不得不面对的问题就是Kafka Producer和Kafka Consumer之间的序列化和反序列化的兼容性,试想对于StringSerializer来说,Kafka Consumer可以顺其自然的采用StringDeserializer,不过对于Company这种专用类型,某个服务使用DemoSerializer进行了序列化之后,那么下游的消费者服务必须也要实现对应的DemoDeserializer。再者,如果上游的Company类型改变,下游也需要跟着重新实现一个新的DemoSerializer,这个后面所面临的难题可想而知。所以,如无特殊需要,笔者不建议使用自定义的序列化和反序列化器;如有业务需要,也要使用通用的Avro、Protobuf、Protostuff等序列化工具包装,尽可能的实现得更加通用且向前后兼容。

题外话,对于Kafka的“深耕者”Confluent来说,还有其自身的一套序列化和反序列化解决方案(io.confluent.kafka.serializer.KafkaAvroSerializer),GitHub上有相关资料,读者如有兴趣可以自行扩展学习。


参考资料
1. protostuff序列化/反序列化


欢迎支持笔者新书:《RabbitMQ实战指南》以及关注微信公众号:Kafka技术专栏。
这里写图片描述

目录
相关文章
|
2月前
|
JSON 数据格式 索引
Python中序列化/反序列化JSON格式的数据
【11月更文挑战第4天】本文介绍了 Python 中使用 `json` 模块进行序列化和反序列化的操作。序列化是指将 Python 对象(如字典、列表)转换为 JSON 字符串,主要使用 `json.dumps` 方法。示例包括基本的字典和列表序列化,以及自定义类的序列化。反序列化则是将 JSON 字符串转换回 Python 对象,使用 `json.loads` 方法。文中还提供了具体的代码示例,展示了如何处理不同类型的 Python 对象。
|
2月前
|
存储 安全 Java
Java编程中的对象序列化与反序列化
【10月更文挑战第22天】在Java的世界里,对象序列化和反序列化是数据持久化和网络传输的关键技术。本文将带你了解如何在Java中实现对象的序列化与反序列化,并探讨其背后的原理。通过实际代码示例,我们将一步步展示如何将复杂数据结构转换为字节流,以及如何将这些字节流还原为Java对象。文章还将讨论在使用序列化时应注意的安全性问题,以确保你的应用程序既高效又安全。
|
3月前
|
存储 Java
Java编程中的对象序列化与反序列化
【10月更文挑战第9天】在Java的世界里,对象序列化是连接数据持久化与网络通信的桥梁。本文将深入探讨Java对象序列化的机制、实践方法及反序列化过程,通过代码示例揭示其背后的原理。从基础概念到高级应用,我们将一步步揭开序列化技术的神秘面纱,让读者能够掌握这一强大工具,以应对数据存储和传输的挑战。
|
3月前
|
存储 安全 Java
Java编程中的对象序列化与反序列化
【10月更文挑战第3天】在Java编程的世界里,对象序列化与反序列化是实现数据持久化和网络传输的关键技术。本文将深入探讨Java序列化的原理、应用场景以及如何通过代码示例实现对象的序列化与反序列化过程。从基础概念到实践操作,我们将一步步揭示这一技术的魅力所在。
|
3月前
|
消息中间件 存储 Java
大数据-58 Kafka 高级特性 消息发送02-自定义序列化器、自定义分区器 Java代码实现
大数据-58 Kafka 高级特性 消息发送02-自定义序列化器、自定义分区器 Java代码实现
82 3
|
3月前
|
消息中间件 存储 分布式计算
大数据-61 Kafka 高级特性 消息消费02-主题与分区 自定义反序列化 拦截器 位移提交 位移管理 重平衡
大数据-61 Kafka 高级特性 消息消费02-主题与分区 自定义反序列化 拦截器 位移提交 位移管理 重平衡
35 1
|
2月前
|
存储 缓存 NoSQL
一篇搞懂!Java对象序列化与反序列化的底层逻辑
本文介绍了Java中的序列化与反序列化,包括基本概念、应用场景、实现方式及注意事项。序列化是将对象转换为字节流,便于存储和传输;反序列化则是将字节流还原为对象。文中详细讲解了实现序列化的步骤,以及常见的反序列化失败原因和最佳实践。通过实例和代码示例,帮助读者更好地理解和应用这一重要技术。
75 0
|
4月前
|
JSON 安全 编译器
扩展类实例的序列化和反序列化
扩展类实例的序列化和反序列化
55 1
|
5月前
|
存储 Java
【IO面试题 四】、介绍一下Java的序列化与反序列化
Java的序列化与反序列化允许对象通过实现Serializable接口转换成字节序列并存储或传输,之后可以通过ObjectInputStream和ObjectOutputStream的方法将这些字节序列恢复成对象。
|
5月前
|
存储 开发框架 .NET
解锁SqlSugar新境界:利用Serialize.Linq实现Lambda表达式灵活序列化与反序列化,赋能动态数据查询新高度!
【8月更文挑战第3天】随着软件开发复杂度提升,数据查询的灵活性变得至关重要。SqlSugar作为一款轻量级、高性能的.NET ORM框架,简化了数据库操作。但在需要跨服务共享查询逻辑时,直接传递Lambda表达式不可行。这时,Serialize.Linq库大显身手,能将Linq表达式序列化为字符串,实现在不同服务间传输查询逻辑。结合使用SqlSugar和Serialize.Linq,不仅能够保持代码清晰,还能实现复杂的动态查询逻辑,极大地增强了应用程序的灵活性和可扩展性。
181 2