文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

kafka序列化器和拦截器怎么自定义使用

2023-07-05 18:27

关注

本篇内容主要讲解“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获取消息后需要反序列化。

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());

自定义序列化

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");    }  }  }

自定义反序列化

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,只需实现它们即可实现拦截,加入拦截器后,生产者会在发送消息之前对消息进行拦截处理,消费者在收到消息之前也会经过拦截器,那么我们就可以在拦截器中加入一些自己需要的逻辑。

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

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序列化器和拦截器怎么自定义使用”有了更深的了解,不妨来实际操作一番吧!这里是编程网网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯