序列化与反序列化_深入理解kafka的序列化和反序列化

kafka内部发送和接收消息的时候&＃xff0c;使用的是byte[]字节数组的方式(RPC底层也是用这种通讯格式)。但是我们在应用层其实可以使用更多的数据类型&＃xff0c;比如int&＃xff0c;short&＃xff0c; long&＃xff0c;String等&＃xff0c;这归功于kafka的序列化和反序列化机制。

在之前的一篇文章springboot集成kafka示例中&＃xff0c;我使用的是kafka原生的StringSerializer序列化方式&＃xff0c;

spring.kafka.producer.key-serializer&＃61;org.apache.kafka.common.serialization.StringSerializerspring.kafka.producer.value-serializer&＃61;org.apache.kafka.common.serialization.StringSerializer

源码如下&＃xff1a;

public class StringSerializer implements Serializer { private String encoding &＃61; "UTF8"; public StringSerializer() { } public void configure(Map configs, Boolean isKey) { String propertyName &＃61; isKey ? "key.serializer.encoding" : "value.serializer.encoding"; Object encodingValue &＃61; configs.get(propertyName); if (encodingValue &＃61;&＃61; null) { encodingValue &＃61; configs.get("serializer.encoding"); } if (encodingValue instanceof String) { this.encoding &＃61; (String)encodingValue; } } public byte[] serialize(String topic, String data) { try { return data &＃61;&＃61; null ? null : data.getBytes(this.encoding); } catch (UnsupportedEncodingException var4) { throw new SerializationException("Error when serializing string to byte[] due to unsupported encoding " &＃43; this.encoding); } } public void close() { }}

其实很简单&＃xff0c;configure方法设置序列化(serialize方法)需要使用的编码&＃xff0c;如果没有设置就使用UTF8格式。这个方法是在生成producer实例的时候被调用的。serialize方法使用的就是String的getBytes把String类型的消息转化为byte字节数组。

反序列呢&＃xff1f;聪明如你应该能想到&＃xff0c;使用new String就可以解决了。源码如下&＃xff1a;

&＃64;Override public String deserialize(String topic, byte[] data) { try { if (data &＃61;&＃61; null) return null; else return new String(data, encoding); } catch (UnsupportedEncodingException e) { throw new SerializationException("Error when deserializing byte[] to string due to unsupported encoding " &＃43; encoding); }}---------------------

是不是简单到爆呢&＃xff1f;

其它的内置序列化组件&＃xff0c;像Double, Integer,Long这些原理都类似&＃xff0c;就不一一分析了。

有时候内置的组件不能满足我们的需要。比如我有个自定义的对象要作为kafka的消息进行收发(把对象转化为json字符串通过String的方式也是一种思路)&＃xff0c;希望能有一个针对我这个对象自定义的序列化和反序列化组件。

我们先定义一个消息对象&＃xff0c;

&＃64;Data&＃64;ToStringpublic class Person { private int id; private String name; private int age;}

然后自定义自己的序列化和反序列化实现类&＃xff0c;

&＃64;Slf4jpublic class PersonDeserializer implements Deserializer { &＃64;Override public void configure(Map map, Boolean b) { } &＃64;Override public Person deserialize(String s, byte[] bytes) { log.info("自定义的反序列化-deserialize"); return JSON.parseObject(bytes, Person.class); } &＃64;Override public void close() { }}&＃64;Slf4jpublic class PersonSerializer implements Serializer { private static Gson gson; static { gson &＃61; new GsonBuilder().create(); } &＃64;Override public void configure(Map map, Boolean b) { log.info("自定义的序列化组件--configure"); } &＃64;Override public byte[] serialize(String s, Person person) { log.info("自定义的序列化组件--serialize"); return JSON.toJSONBytes(person); } &＃64;Override public void close() { log.info("自定义的序列化组件--close"); }}

代码一看就明白&＃xff0c;其实核心就是利用fastjson的toJSONBytes把对象转化为byte数组。

然后我们在配置里指定使用我们自己的序列化和反序列化实现类&＃xff0c;

spring.kafka.producer.key-serializer&＃61;org.apache.kafka.common.serialization.StringSerializerspring.kafka.producer.value-serializer&＃61;com.ponymaggie.github.kafka.serializer.PersonSerializerspring.kafka.consumer.key-deserializer&＃61;org.apache.kafka.common.serialization.StringDeserializerspring.kafka.consumer.value-deserializer&＃61;com.ponymaggie.github.kafka.serializer.PersonDeserializer测试

启动springboot项目&＃xff0c;通过日志可以看出消息的收发都是正常的。

2019-08-15 20:06:34.251 INFO 16676 --- [ main] c.p.github.kafka.producer.KafkaSender : &＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43; message &＃61; {"id":1000,"name":"小明