java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > kafka序列化器拦截器

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

作者:刘牌

这篇文章主要为大家介绍了kafka序列化器和拦截器使用示例详解,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪

介绍

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

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

序列化器

数据在网络中传输是以字节流的形式进行传输,在生产者端发送消息需要先进行序列化,消费者端进行反序列化,序列化的方式有很多,比如jdk,json,protobuf,kryo,hessian,avro等等,在大数据量的传输中,序列化和反序列化的效率对吞吐量有一定的影响,kafka提供了许多序列化和反序列化器,如StringDeserializerStringSerializer,如果我们需要自定义一个序列化和反序列化器,那么实现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中消费者和生产者都有拦截器,分别为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 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序列化器拦截器的资料请关注脚本之家其它相关文章!

您可能感兴趣的文章:
阅读全文