一文详解kafka序列化器和 *
作者:刘牌 发布时间:2023-06-18 01:06:08
介绍
本篇主要介绍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
来源:https://juejin.cn/post/7215640327632519227
猜你喜欢
- 字符串操作在任意编程语言的日常编程中都随处可见,今天来汇总一下 C# 中关于字符串的一些你可能遗忘或遗漏的知识点。逐字字符串在普通字符串中,
- 话不多说,先上图 &n
- 释放公平锁(基于JDK1.7.0_40)1. unlock()unlock()在ReentrantLock.java中实现的,源码如下:pu
- 使用mybatis-plus自动生成了5个模块(xml/bean/mapper/service/controller)的代码,这里练习一下m
- 本文实例为大家分享了C# Winform实现进度条显示的具体代码,供大家参考,具体内容如下创建一个窗体,命名为StartForm添加一个ti
- java简单模拟微信抢红包功能,本例发100元红包,有10个人抢,为了尽可能的公平,每个人的红包金额都要随机(保证结果的不确定性,本例抢红包
- spring cloud gateway获取请求的真实地址在使用spring cloud gateway的时候,路由一般配置为服务名例如 l
- 一、解决的痛点 1、免搭建后端开发环境。 &n
- 本文的主要内容包括在下图,下面来一起看看吧。1、&和&&的区别2、switchswitch语句能否作用在byte,能
- 本文实例讲述了C#实现3步手动建DataGridView的方法。分享给大家供大家参考。具体如下:DataView dvCmControlIn
- mybatis中的#和$的区别1. #将传入的数据都当成一个字符串,会对自动传入的数据加一个双引号。如:order by #user_id#
- 前言最近在优化自己之前基于Spring AOP的统一响应体的实现方案。什么是统一响应体呢?在目前的前后端分离架构下,后端主要是一个RESTf
- Java Benchmark 基准测试的实例详解import java.util.Arrays; import java.util.conc
- Java序列化是将一个对象编码成一个字节流,反序列化将字节流编码转换成一个对象。 序列化是
- filter自定义过滤器 增加了 对验证码的校验package com.youxiong.filter;import com.y
- 系统参数系统级全局变量,该参数在程序中任何位置都可以访问到。优先级最高,覆盖程序中同名配置。系统参数的标准格式为:-Dargname=arg
- 一、什么是JSONJSON(JavaScript Object Notation)是一种基于JavaScript语法子集的开放标准数据交换格
- 背景介绍公司最近做分库分表业务,接入了 Sharding JDBC,接入完成后,回归测试时发现好几个 SQL 执行报错,关键这几个表都还不是
- 一、 WillPopScope用法WillPopScope本质是一个widget用于拦截物理按键返回事件(Android的物理返回键和iOS
- Springboot 内置tomcat禁止不安全HTTP方法1、在tomcat的web.xml中可以配置如下内容让tomcat禁止不安全的H