一文详解kafka序列化器和拦截器
作者：刘牌
介绍
本篇主要介绍kafka的拦截器和序列化器，序列化器是和数据在网络中的传输有关，数据在网络中的传输为字节流，所以生产者在发送时需要将其序列化为字节流，消费者收到消息时，需要将字节流反序列化为我们能够识别的对象，我们不难看出，这就是RPC通信，kafka中实现了很多自定义协议，我们知道，在RPC通信中，只有生产者和消费者的协议一样，才能相互传输和解析数据，在使用HTTP时，我们就不用去关注协议本身，因为HTTP是TCP的上层建筑，它自己实现了一套协议，我们不用去关注，但是使用RPC，我们是面向TCP编程，所以自然得约定和实现自己的协议，而序列化就是这过程中很重要的一部分。
拦截器是一个随处可见的词，基本上很多框架中都有拦截器机制，它的作用主要是对请求进行拦截，我们可以对请求进行过滤和处理，以达到业务目的，比如Spring中有HandlerInterceptor拦截器，在kafka种也有拦截器，我们可以自定义拦截器，对消息进行拦截，比如某些异常消息我们不需要发送，那么就将其拦截下来。
序列化器
数据在网络中传输是以字节流的形式进行传输，在生产者端发送消息需要先进行序列化，消费者端进行反序列化，序列化的方式有很多，比如jdk，json，protobuf，kryo，hessian，avro等等，在大数据量的传输中，序列化和反序列化的效率对吞吐量有一定的影响，kafka提供了许多序列化和反序列化器，如StringDeserializer和StringSerializer，如果我们需要自定义一个序列化和反序列化器，那么实现Serializer,Deserializer接口即可。
如下，kafka生产者在发送消息到broker之前需要序列化，消费者从broker获取消息后需要反序列化。
设置序列化和反序列化
生产者端设置序列化
//序列化
props.put("key.serializer", StringSerializer.class.getName());
props.put("value.serializer", StringSerializer.class.getName());
消费者端设置反序列化
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());
自定义序列化
/**
* 功能说明： JSON序列化
* <p>
* Original @Author: steakliu , 2022-11-02 15:14
*/
public class JsonSerializer<T> implements Serializer<T> {
@Override
public byte[] serialize(String topic, T obj) {
try {
return obj == null ? null : JSON.toJSONBytes(obj);
}catch (Exception e){
throw new SerializationException("json serializing exception");
}
}
}
自定义反序列化
/**
* 功能说明：JSON反序列化
* <p>
* Original @Author: steakliu-刘牌, 2022-11-11 09:38
*/
public class JsonDeserializer<T> implements Deserializer<T> {
@Override
public T deserialize(String topic, byte[] data) {
return (T) JSON.parse(data);
}
}
如上简单的使用fastjson作为序列化和反序列化工具，演示了自定义kafka的序列化和反序列化机制，我们可以根据实际情况来设计不同的序列化反序列化机制，当然，不会是像上面这些简单，如果使用spring，那么spring提供了JSON序列化和反序列化器直接使用。
思考
虽然我们可以自定义序列化和反序列化器，但是自定义序列化和反序列化器在使用上也要保持一些一致，也就是说生产者和消费者要保持使用一种类型的序列化机制，不然会出现消息转换问题，如果我们以kafka的方式向别人提供服务，那么他们就需要使用我们的制定的序列化方式，所以这可能就存在一定的耦合，如果使用Kafka的String序列化和反序列化机制，因为是它是默认方式并且是字符串，通用性比较好，所以就不用去考虑序列化和反序列化，直接拿到字符串转为对象，再进行业务处理，使用自定义序列化的话，就直接拿到序列化后的对象，不用进行字符串转对象操作。
在实际场景中，我们可以根据自己的业务来使用何种序列化方式，没有最好的，只有合适的。
拦截器
kafka中消费者和生产者都有拦截器，分别为ConsumerInterceptor和ProducerInterceptor，只需实现它们即可实现拦截，加入拦截器后，生产者会在发送消息之前对消息进行拦截处理，消费者在收到消息之前也会经过拦截器，那么我们就可以在拦截器中加入一些自己需要的逻辑。
如下消费者拦截器对消息进行拦截，如果有异常消息，则对异常消息进行处理，只要需要对消息进行处理，监控等，都可以使用拦截器。
/**
* 功能说明： 消费者拦截器
* <p>
* Original @Author: steakliu-刘牌, 2023-03-15 10:17
*/
public class MyConsumerInterceptor implements ConsumerInterceptor<String, Message> {
@Override
public ConsumerRecords<String, Message> onConsume(ConsumerRecords<String, Message> records) {
long currentTimeMillis = System.currentTimeMillis();
records.forEach(record -> {
if ("消息异常".equals(record.value().getMessageText())) {
//处理异常消息
this.handleMsg(record);
}
});
return records;
}
private void handleMsg(ConsumerRecord<String, Message> record) {
//处理异常消息
}
@Override
public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {}
@Override
public void close() {}
@Override
public void configure(Map<String, ?> configs) { }
}
拦截器可以有多个，如果设置多个拦截器，那么就形成一个拦截器链，一个一个地执行。
下面是使用spring-kafka时所配置的拦截器和序列化器的基本配置。
spring:
kafka:
bootstrap-servers: 127.0.0.1:9092
consumer:
# 反序列化器
key-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
properties:
# 拦截器
interceptor:
classes: com.steakliu.kafka.interceptor.MyConsumerInterceptor
spring:
json:
trusted:
packages: '*'
producer:
key-serializer: org.springframework.kafka.support.serializer.JsonSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
properties:
# 拦截器
interceptor:
classes: com.steakliu.kafka.interceptor.MyProducerInterceptor,com.steakliu.kafka.interceptor.MyProducerInterceptor2
总结
对于拦截器和序列化器，我们上面作了简单地描述和示例，对于它们，可能我们都不怎么去去用甚至没有用过，但是还是很有必要去了解的，了解它的设计和思想，在一些特殊的场景可能会用到。
以上就是一文详解kafka序列化器和拦截器的详细内容，更多关于kafka序列化器拦截器的资料请关注脚本之家其它相关文章！