0


kafka消息的序列化与反序列化

一、前言

在使用kafka发送消息时,producer端需要序列化,在大多数场景中, 需要传输的是与业务规则相关的复杂类型, 这就需要自定义数据结构。 Avro是一种序列化框架, 使用JSON来定义schema, sh cema由原始类型(null, boolean, int, long, float, double, bytes, string) 和复杂类型 (record, enum, array, map, union, fixed) 组成, schema文件以.avsc结尾, 表示avro schema

有2种序列化方式

  • 二进制方式:也就是Specific方式, 定义好schema a svc文件后, 使用编译器(avro-tools.jar) 编译生成相关语言(java) 的业务类, 类中会嵌入JSON schema ·
  • JSON方式:也就是Generic方式, 在代码中动态加载schema a svc文件, 将FieldName-Field Value, 以Map<K, V>的方式存储

序列化后的数据号schema和data同时存在的, 如下图、
在这里插入图片描述

二、序列化与反序列化实现

1、序列化(producer)

props.put("key.serializer",StringSerializer.class.getName());
props.put("value.serializer",StringSerializer.class.getName());

2、反序列化(consumer)

props.put("key.deserializer",StringDeserializer.class.getName());
props.put("value.deserializer",StringDeserializer.class.getName());

如果是flink,也可以用:

KafkaSourceBuilder 类提供了两个方法来反序列数据,分别是 setDeserializer 和 setValueOnlyDeserializer
从名字上就应该可以看出这两者的区别,前者是反序列化完整的 ConsumerRecord,后者只反序列化 ConsumerRecord 的 value.

如果你想要获取 kafka 的元数据信息选择实现 KafkaDeserializationSchema 接口就可以了,KafkaDeserializationSchema 接口还有 4 个静态方法,其中的 of 方法就是用来反序列化 ConsumerRecord 的,剩下的 3 个 valueOnly 是用来反序列化 kafka 消息中的 value 的.

KafkaSource<String> source =KafkaSource.<String>builder().setBootstrapServers("ip").setTopics("web-topic").setGroupId("web-group").setStartingOffsets(OffsetsInitializer.earliest())//低版本jdk报错,改成jdk8.setValueOnlyDeserializer(newSimpleStringSchema()).build();
标签: kafka java 分布式

本文转载自: https://blog.csdn.net/paulwang040/article/details/128044775
版权归原作者 菠菜很好吃 所有, 如有侵权,请联系我们删除。

“kafka消息的序列化与反序列化”的评论:

还没有评论