开发者

一文详解kafka序列化器和拦截器

目录
  • 介绍
  • 序列化器
    • 设置序列化和反序列化
    • 自定义序列化
    • 自定义反序列化
    • 思考
  • 拦截器
    • 总结

      介绍

      本篇主要介绍kafka的拦截器和序列化器,序列化器是和数据在网络中的传输有关,数据在网络中的传输为字节流,所以生产者在发送时需要将其序列化为字节流,消费者收到消息时,需要将字节流反序列化为我们能够识别的对象,我们不难看出,这就是RPC通信,kafka中实现了很多自定义协议,我们知道,在RPC通信中,只有生产者和消费者的协议一样,才能相互传输和解析数据,在使用HTTP时,我们就不用去关注协议本身,因为HTTP是TCP的上层建筑,它自己实现了一套协议,我们不用去关注,但是使用RPC,我们是面向TCP编程,所以自然得约定和实现自己的协议,而序列化就是这过程中很重要的一部分。

      拦截器是一个随处可见的词,基本上很多框架中都有拦截器机制,它的作用主要是对请求进行拦截,我们可以对请求进行过滤和处理,以达到业务目的,比如Spring中有HandlerInterceptor拦截器,在kafka种也有拦截器,我们可以自定义拦截器,对消息进行拦截,比如某些异常消息我们不python需要发送,那么就将其拦截下来。

      序列化器

      数据在网络中传输是以字节流的形式进行传输,在生产者端发送消息需要先进行序列化,消费者端进行反序列化,序列化的方式有很多,比如jdk,json,protobuf,kryo,hessian,avro等等,在大数据量的传输中,序列化和反序列化的效率对吞吐量有一定的影响,kafka提供了许多序列化和反序列化器,如StringDeserializerStringSerializer,如果我们需要自定义一个序列化和反序列化器,那么实现Serializer,Deserializer接口即可。

      如下,kafka生产者在发送消息到broker之前android需要序列化,消费者从broker获取消息后需要反序列化。

      一文详解kafka序列化器和拦截器

      设置序列化和反序列化

      生产者端设置序列化

      //序列化
      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 : JSjsON.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序列化和反序列化器直接使用。

      思考

      虽然我们可以自定义序列化和反序列化器,但是自定义序列化和反序列js化器在使用上也要保持一些一致,也就是说生产者和消费者要保持使用一种类型的序列化机制,不然会出现消息转换问题,如果我们以kafka的方式向别人提供服务,那么他们就需要使用我们的制定的序列化方式,所以这可能就存在一定的耦合,如果使用Kafka的String序列化和反序列化机制,因为是它是默认方式并且是字符串,通用性比较好,所以就不用去考虑序列化和反序列化,直接拿到字符串转为对象,再进行业务处理,使用自定义序列化的话,就直接拿到序列化后的对象,不用进行字符串转对象操作。

      在实际场景中,我们可以根据自己的业务来使用何种序列化方式,没有最好的,只有合适的。

      拦截器

      kafka中消费者和生产者都有拦截器,分别为ConsumerInterceptorProducerInterceptor,只需实现它们即可实现拦截,加入拦截器后,生产者会在发送消息之前对消息进行拦截处理,消费者在收到消息之前也会经过拦截器,那么我们就可以在拦截器中加入一些自己需要的逻辑。

      如下消费者拦截器对消息进行拦截,如果有异常消息,则对异常消息进行处理,只要需要对消息进行处理,监控等,都可以使用拦截器。

      /**
       * 功能说明: 消费者拦截器
       * <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 开发者_Python教程close() {}
        @Override
        public void configure(Map<String, ?> configs) { }
      }
      

      拦截器可以有多个,如果设置多个拦截器,那么就形成一个拦截器链,一个一个地执行。

      下面是使用spring-kafka时所配置的拦截器和序列化器的基本配置。

      spring:
        kafka:
          bootstrap-servers: 127.0.0.1:9092
          consumer:
            # 反序列化器
            key-deserializer: www.devze.comorg.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序列化器拦截器的资料请关注我们其它相关文章!

      0

      上一篇:

      下一篇:

      精彩评论

      暂无评论...
      验证码 换一张
      取 消

      最新开发

      开发排行榜