简介
kafka内部发送和接收消息的时候,使用的是byte[]字节数组的方式(RPC底层也是用这种通讯格式)。但是我们在应用层其实可以使用更多的数据类型,比如int,short, long,String等,这归功于kafka的序列化和反序列化机制。
基本原理分析
在之前的一篇文章springboot集成kafka示例中,我使用的是kafka原生的StringSerializer序列化方式,
1 |
|
源码如下:
1 |
|
其实很简单,configure方法设置序列化(serialize方法)需要使用的编码,如果没有设置就使用UTF8格式。这个方法是在生成producer实例的时候被调用的。serialize方法使用的就是String的getBytes把String类型的消息转化为byte字节数组。
反序列呢?聪明如你应该能想到,使用new String就可以解决了。源码如下,
1 |
|
是不是简单到爆呢?
其它的内置序列化组件,像Double, Integer,Long这些原理都类似,就不一一分析了。
自定义序列化组件
有时候内置的组件不能满足我们的需要。比如我有个自定义的对象要作为kafka的消息进行收发(把对象转化为json字符串通过String的方式也是一种思路),希望能有一个针对我这个对象自定义的序列化和反序列化组件。
我们先定义一个消息对象,
1 |
|
然后自定义自己的序列化和反序列化实现类,
1 |
|
1 |
|
代码一看就明白,其实核心就是利用fastjson的toJSONBytes把对象转化为byte数组。
然后我们在配置里指定使用我们自己的序列化和反序列化实现类,
1 |
|
测试
本地kafka和zk环境搭建可以参考我之前的一篇文章:
http://www.machengyu.net/arch/2019/07/29/kafka-springboot.html
启动springboot项目,通过日志可以看出消息的收发都是正常的。
1 |
|
源码地址:
https://github.com/pony-maggie/springboot-kafka-serialization-demo