上一篇【第12篇】Kafka拦截器源码解析——你不知道的消息预处理神器下一篇【第14篇】Kafka分区器源码解析——消息去哪个分区有学问摘要Kafka只认识字节数组这是它高性能的秘密之一——不关心你发的到底是什么只管传输byte[]。但作为开发者我们需要把Java对象变成byte[]才能发送这个过程就叫序列化。Kafka提供了Serialzier/Deserializer接口以及String、Integer、ByteArray等内置实现但实际开发中Protobuf和Avro才是主流选择。本文将深度解析Serializer接口设计剖析内置序列化器源码手把手教你写出生产级的Protobuf和Avro序列化器。读完这篇自定义Serializer不再是让你头秃的事情。一、为什么需要序列化器——Kafka只认识byte[]先搞清楚一个事实Kafka的Broker不关心你发的消息内容是什么。【Kafka的数据抽象层】 Java对象 byte[] byte[] Java对象 ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ Order │ ──序列化──►│ 0x00 0A..│ │ 0x00 0A..│ ──反序列化──►│ Order │ │{id, name,│ │{payload} │ ──►Kafka──►│{payload} │ │{id, name,│ │ amount} │ │ │ │ │ │ amount} │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ ▲ ▲ ▲ ▲ │ │ │ │ Producer Serializer Deserializer Consumer消息在KafkaProducer中的流转位置很重要。在发送流程中序列化发生在消息进入RecordAccumulator之前KafkaProducer.send() 核心流程 ┌─────────────────────────────────────────────────────────────────┐ │ │ │ ProducerRecord ──► Interceptors.onSend() ──► Serializer.serialize() │ │ │ │ ▼ │ │ byte[] key, value │ │ │ │ │ ▼ │ │ Partitioner.partition() │ │ │ │ │ ▼ │ │ RecordAccumulator.append()│ │ │ └─────────────────────────────────────────────────────────────────┘注意序列化在分区选择之前原因很简单——如果用了Key Hash分区策略你得先把Key序列化成byte[]才能计算Hash值。二、Serializer接口——简单到不可思议Kafka的Serializer接口只定义了三个方法简单得让人怀疑是不是少写了什么publicinterfaceSerializerTextendsCloseable{/** * 配置方法在序列化操作之前调用 * param configs 配置参数如编码格式等 * param isKey 标识这个Serializer是用于Key还是Value */voidconfigure(MapString,?configs,booleanisKey);/** * 核心方法将Java对象序列化为byte数组 * param topic 发送到的Topic名称 * param data 需要序列化的Java对象 * return 序列化后的byte数组 */byte[]serialize(Stringtopic,Tdata);/** * 关闭资源大多数实现为空 */Overridevoidclose();}三个方法的职责非常清晰configure()——初始化配置serialize()——执行序列化close()——清理资源。配合对应的Deserializer接口publicinterfaceDeserializerTextendsCloseable{voidconfigure(MapString,?configs,booleanisKey);Tdeserialize(Stringtopic,byte[]data);Overridevoidclose();}序列化和反序列化成对出现这就像锁匠——你配了一把钥匙Serializer还得配一把对应的锁Deserializer。三、内置序列化器源码解析——Kafka自带的开箱即用方案3.1 StringSerializer——最常用的序列化器publicclassStringSerializerimplementsSerializerString{privateStringencodingUTF8;// 默认UTF8编码Overridepublicvoidconfigure(MapString,?configs,booleanisKey){// 从配置中读取编码格式StringpropertyNameisKey?key.serializer.encoding:value.serializer.encoding;ObjectencodingValueconfigs.get(propertyName);if(encodingValueinstanceofString)encoding(String)encodingValue;}Overridepublicbyte[]serialize(Stringtopic,Stringdata){try{if(datanull)returnnull;// null直接返回null注意后续null处理elsereturndata.getBytes(encoding);// 按指定编码转byte[]}catch(UnsupportedEncodingExceptione){thrownewSerializationException(Error serializing string,e);}}}核心逻辑就是data.getBytes(encoding)简单直接。但注意null的处理——如果传入null返回null而不是空数组。这意味着上层调用者必须处理好null的情况。3.2 ByteArraySerializer——不序列化的序列化器publicclassByteArraySerializerimplementsSerializerbyte[]{Overridepublicbyte[]serialize(Stringtopic,byte[]data){returndata;// 直接原样返回}}如果你发的已经是byte[]比如用Protobuf在外面序列化了那这个序列化器就是你的不二之选——不做任何转换性能开销为零。3.3 IntegerSerializer——数字的序列化器publicclassIntegerSerializerimplementsSerializerInteger{Overridepublicbyte[]serialize(Stringtopic,Integerdata){if(datanull)returnnull;// 4字节的int转byte数组returnnewbyte[]{(byte)(data24),// 高8位(byte)(data16),(byte)(data8),data.byteValue()// 低8位};}}使用位运算实现大端序Big-Endian编码网络传输的标准做法。3.4 内置序列化器对比序列化器适用类型序列化方式大小一个int适用场景StringSerializerStringUTF-8编码可变JSON格式消息ByteArraySerializerbyte[]原样返回原始大小已预序列化数据ByteBufferSerializerByteBufferbuffer.array()原始大小NIO缓冲区IntegerSerializerInteger大端4字节4字节数值型KeyLongSerializerLong大端8字节8字节时间戳KeyDoubleSerializerDoubleIEEE 7548字节浮点数ShortSerializerShort大端2字节2字节短整数四、实战自定义Protobuf序列化器内置序列化器能用但生产环境通常用Protobuf或Avro。下面我们来写一个生产级别的Protobuf序列化器。4.1 定义Protobuf消息// order.proto syntax proto3; message Order { int64 order_id 1; string user_id 2; double amount 3; int64 timestamp 4; }4.2 实现通用Protobuf序列化器importcom.google.protobuf.MessageLite;importcom.google.protobuf.Parser;importorg.apache.kafka.common.serialization.Serializer;importorg.apache.kafka.common.serialization.Deserializer;/** * 通用Protobuf序列化器 * 因为Protobuf生成的类都实现了MessageLite接口 * 所以一个序列化器可以处理所有Protobuf消息 */publicclassProtobufSerializerTextendsMessageLiteimplementsSerializerT{Overridepublicbyte[]serialize(Stringtopic,Tdata){if(datanull){returnnull;}returndata.toByteArray();// Protobuf自带序列化方法}Overridepublicvoidclose(){// 无需释放资源}}/** * 通用Protobuf反序列化器 */publicclassProtobufDeserializerTextendsMessageLiteimplementsDeserializerT{privateParserTparser;publicProtobufDeserializer(ParserTparser){this.parserparser;}OverridepublicTdeserialize(Stringtopic,byte[]data){if(datanull){returnnull;}try{returnparser.parseFrom(data);// Protobuf自带反序列化}catch(InvalidProtocolBufferExceptione){thrownewSerializationException(Failed to deserialize,e);}}Overridepublicvoidclose(){// 无需释放资源}}4.3 使用示例// Producer端配置PropertiespropsnewProperties();props.put(key.serializer,org.apache.kafka.common.serialization.StringSerializer);props.put(value.serializer,com.example.ProtobufSerializer);// 使用自定义序列化器KafkaProducerString,OrderproducernewKafkaProducer(props);OrderorderOrder.newBuilder().setOrderId(1001L).setUserId(user_123).setAmount(99.99).setTimestamp(System.currentTimeMillis()).build();producer.send(newProducerRecord(orders,order.getUserId(),order));【Protobuf序列化 vs JSON序列化 对比】 Order对象 序列化结果 ┌─────────────────┐ ┌──────────────────────┐ │ orderId: 1001 │ Protobuf │ 08 E9 07 12 08 ... │ 约 25 bytes │ userId: user_123│ ──────────────► │ │ │ amount: 99.99 │ │ │ │ timestamp: ... │ JSON │ {orderId:1001, │ 约 120 bytes └─────────────────┘ ──────────────► │ userId:user_123,│ │ amount:99.99, │ │ timestamp:...} │ └──────────────────────┘ Protobuf序列化结果比JSON小约80%因为使用Varint变长编码五、实战Avro序列化器带Schema RegistryAvro比Protobuf更进一步——它把Schema模式定义和数据分离Schema可以独立管理数据本身不携带字段名更省空间。5.1 Avro序列化器实现importorg.apache.avro.Schema;importorg.apache.avro.generic.GenericRecord;importorg.apache.avro.io.*;importorg.apache.avro.specific.SpecificDatumWriter;importorg.apache.avro.specific.SpecificDatumReader;/** * Avro序列化器 * 使用Specific Record模式代码生成 */publicclassAvroSerializerTextendsSpecificRecordimplementsSerializerT{Overridepublicbyte[]serialize(Stringtopic,Tdata){if(datanull)returnnull;try{ByteArrayOutputStreamoutnewByteArrayOutputStream();BinaryEncoderencoderEncoderFactory.get().binaryEncoder(out,null);SpecificDatumWriterTwriternewSpecificDatumWriter(data.getSchema());// Avro二进制编码先写Schema指纹再写数据byte[]schemaFingerprintdata.getSchema().toString().getBytes();out.write(intToByteArray(schemaFingerprint.length));// 4字节长度out.write(schemaFingerprint);// Schema内容writer.write(data,encoder);// 数据内容encoder.flush();returnout.toByteArray();}catch(IOExceptione){thrownewSerializationException(Avro serialization failed,e);}}privatebyte[]intToByteArray(intvalue){returnnewbyte[]{(byte)(value24),(byte)(value16),(byte)(value8),(byte)value};}}/** * Avro反序列化器 */publicclassAvroDeserializerTextendsSpecificRecordimplementsDeserializerT{privatefinalClassTtargetType;publicAvroDeserializer(ClassTtargetType){this.targetTypetargetType;}OverridepublicTdeserialize(Stringtopic,byte[]data){if(datanull)returnnull;try{// 解析SchemaintschemaLenByteBuffer.wrap(data,0,4).getInt();StringschemaStrnewString(data,4,schemaLen);SchemaschemanewSchema.Parser().parse(schemaStr);// 解析数据ByteArrayInputStreaminnewByteArrayInputStream(data,4schemaLen,data.length-4-schemaLen);BinaryDecoderdecoderDecoderFactory.get().binaryDecoder(in,null);SpecificDatumReaderTreadernewSpecificDatumReader(schema);// 创建新实例注意Avro反序列化不回填原对象TresulttargetType.getDeclaredConstructor().newInstance();reader.read(result,decoder);returnresult;}catch(Exceptione){thrownewSerializationException(Avro deserialization failed,e);}}}5.2 Avro Schema定义{type:record,name:OrderAvro,namespace:com.example.avro,fields:[{name:orderId,type:long},{name:userId,type:string},{name:amount,type:double},{name:timestamp,type:long}]}六、序列化方案终极对比维度JSONProtobufAvro序列化大小大文本格式小二进制Varint最小Schema分离可读性✅ 人类可读❌ 二进制❌ 二进制Schema演化❌ 无约束✅ 向前兼容✅✅ 最佳独立Schema跨语言支持✅ 所有语言✅ 多语言✅ 多语言序列化速度慢文本解析快快需要Schema❌ 不需要编译时检查运行时检查适用场景调试开发、日志内部微服务数据管道、流处理与Kafka集成度一般好✅✅ 最佳Confluent原生七、生产环境最佳实践7.1 序列化器选择决策树┌─── 是否需要跨语言互操作 │ ┌──否──┤ ┌── 性能要求极高 │ │ │ │ └──是──► 消息量级 ──► 少量 ──► JSON简单可读 │ │ │ └── 海量 ──► Schema管理能力 │ │ └── JSON ┌─── 有 ──► Avro Schema Registry │ └─── 无 ──► Protobuf7.2 常见踩坑Key和Value序列化器必须成对配置Producer配了自定义SerializerConsumer必须配对应的Deserializernull值处理Kafka允许消息的Key或Value为null但自定义序列化器必须能处理null输入序列化异常传播serialize()中的异常要包装成SerializationException抛出不要复用序列化器状态如果serialize()里用了成员变量确保它是线程安全的——因为KafkaProducer是线程安全的一个Producer实例可能被多个线程共享本篇小结本文带你彻底搞懂了Kafka的序列化机制设计哲学Kafka只认识byte[]序列化/反序列化完全由客户端负责服务端什么都不管——这也是Kafka高性能的秘诀之一接口设计Serializer/Deserializer接口简单到只有三个方法开箱即用的内置实现覆盖了所有基本类型实战重点Protobuf序列化器适合内部微服务数据小、速度快Avro序列化器配合Schema Registry适合数据管道Schema演化能力最强选型建议调试用JSON内部微服务用Protobuf跨组织数据管道用Avro下一篇我们来聊分区器看看消息是怎么决定去哪个分区的上一篇【第12篇】Kafka拦截器源码解析——你不知道的消息预处理神器下一篇【第14篇】Kafka分区器源码解析——消息去哪个分区有学问
【Kafka源码解读和使用指南】第13篇：Kafka序列化器深度解析——自定义Serializer不再是难题
上一篇【第12篇】Kafka拦截器源码解析——你不知道的消息预处理神器下一篇【第14篇】Kafka分区器源码解析——消息去哪个分区有学问摘要Kafka只认识字节数组这是它高性能的秘密之一——不关心你发的到底是什么只管传输byte[]。但作为开发者我们需要把Java对象变成byte[]才能发送这个过程就叫序列化。Kafka提供了Serialzier/Deserializer接口以及String、Integer、ByteArray等内置实现但实际开发中Protobuf和Avro才是主流选择。本文将深度解析Serializer接口设计剖析内置序列化器源码手把手教你写出生产级的Protobuf和Avro序列化器。读完这篇自定义Serializer不再是让你头秃的事情。一、为什么需要序列化器——Kafka只认识byte[]先搞清楚一个事实Kafka的Broker不关心你发的消息内容是什么。【Kafka的数据抽象层】 Java对象 byte[] byte[] Java对象 ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ Order │ ──序列化──►│ 0x00 0A..│ │ 0x00 0A..│ ──反序列化──►│ Order │ │{id, name,│ │{payload} │ ──►Kafka──►│{payload} │ │{id, name,│ │ amount} │ │ │ │ │ │ amount} │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ ▲ ▲ ▲ ▲ │ │ │ │ Producer Serializer Deserializer Consumer消息在KafkaProducer中的流转位置很重要。在发送流程中序列化发生在消息进入RecordAccumulator之前KafkaProducer.send() 核心流程 ┌─────────────────────────────────────────────────────────────────┐ │ │ │ ProducerRecord ──► Interceptors.onSend() ──► Serializer.serialize() │ │ │ │ ▼ │ │ byte[] key, value │ │ │ │ │ ▼ │ │ Partitioner.partition() │ │ │ │ │ ▼ │ │ RecordAccumulator.append()│ │ │ └─────────────────────────────────────────────────────────────────┘注意序列化在分区选择之前原因很简单——如果用了Key Hash分区策略你得先把Key序列化成byte[]才能计算Hash值。二、Serializer接口——简单到不可思议Kafka的Serializer接口只定义了三个方法简单得让人怀疑是不是少写了什么publicinterfaceSerializerTextendsCloseable{/** * 配置方法在序列化操作之前调用 * param configs 配置参数如编码格式等 * param isKey 标识这个Serializer是用于Key还是Value */voidconfigure(MapString,?configs,booleanisKey);/** * 核心方法将Java对象序列化为byte数组 * param topic 发送到的Topic名称 * param data 需要序列化的Java对象 * return 序列化后的byte数组 */byte[]serialize(Stringtopic,Tdata);/** * 关闭资源大多数实现为空 */Overridevoidclose();}三个方法的职责非常清晰configure()——初始化配置serialize()——执行序列化close()——清理资源。配合对应的Deserializer接口publicinterfaceDeserializerTextendsCloseable{voidconfigure(MapString,?configs,booleanisKey);Tdeserialize(Stringtopic,byte[]data);Overridevoidclose();}序列化和反序列化成对出现这就像锁匠——你配了一把钥匙Serializer还得配一把对应的锁Deserializer。三、内置序列化器源码解析——Kafka自带的开箱即用方案3.1 StringSerializer——最常用的序列化器publicclassStringSerializerimplementsSerializerString{privateStringencodingUTF8;// 默认UTF8编码Overridepublicvoidconfigure(MapString,?configs,booleanisKey){// 从配置中读取编码格式StringpropertyNameisKey?key.serializer.encoding:value.serializer.encoding;ObjectencodingValueconfigs.get(propertyName);if(encodingValueinstanceofString)encoding(String)encodingValue;}Overridepublicbyte[]serialize(Stringtopic,Stringdata){try{if(datanull)returnnull;// null直接返回null注意后续null处理elsereturndata.getBytes(encoding);// 按指定编码转byte[]}catch(UnsupportedEncodingExceptione){thrownewSerializationException(Error serializing string,e);}}}核心逻辑就是data.getBytes(encoding)简单直接。但注意null的处理——如果传入null返回null而不是空数组。这意味着上层调用者必须处理好null的情况。3.2 ByteArraySerializer——不序列化的序列化器publicclassByteArraySerializerimplementsSerializerbyte[]{Overridepublicbyte[]serialize(Stringtopic,byte[]data){returndata;// 直接原样返回}}如果你发的已经是byte[]比如用Protobuf在外面序列化了那这个序列化器就是你的不二之选——不做任何转换性能开销为零。3.3 IntegerSerializer——数字的序列化器publicclassIntegerSerializerimplementsSerializerInteger{Overridepublicbyte[]serialize(Stringtopic,Integerdata){if(datanull)returnnull;// 4字节的int转byte数组returnnewbyte[]{(byte)(data24),// 高8位(byte)(data16),(byte)(data8),data.byteValue()// 低8位};}}使用位运算实现大端序Big-Endian编码网络传输的标准做法。3.4 内置序列化器对比序列化器适用类型序列化方式大小一个int适用场景StringSerializerStringUTF-8编码可变JSON格式消息ByteArraySerializerbyte[]原样返回原始大小已预序列化数据ByteBufferSerializerByteBufferbuffer.array()原始大小NIO缓冲区IntegerSerializerInteger大端4字节4字节数值型KeyLongSerializerLong大端8字节8字节时间戳KeyDoubleSerializerDoubleIEEE 7548字节浮点数ShortSerializerShort大端2字节2字节短整数四、实战自定义Protobuf序列化器内置序列化器能用但生产环境通常用Protobuf或Avro。下面我们来写一个生产级别的Protobuf序列化器。4.1 定义Protobuf消息// order.proto syntax proto3; message Order { int64 order_id 1; string user_id 2; double amount 3; int64 timestamp 4; }4.2 实现通用Protobuf序列化器importcom.google.protobuf.MessageLite;importcom.google.protobuf.Parser;importorg.apache.kafka.common.serialization.Serializer;importorg.apache.kafka.common.serialization.Deserializer;/** * 通用Protobuf序列化器 * 因为Protobuf生成的类都实现了MessageLite接口 * 所以一个序列化器可以处理所有Protobuf消息 */publicclassProtobufSerializerTextendsMessageLiteimplementsSerializerT{Overridepublicbyte[]serialize(Stringtopic,Tdata){if(datanull){returnnull;}returndata.toByteArray();// Protobuf自带序列化方法}Overridepublicvoidclose(){// 无需释放资源}}/** * 通用Protobuf反序列化器 */publicclassProtobufDeserializerTextendsMessageLiteimplementsDeserializerT{privateParserTparser;publicProtobufDeserializer(ParserTparser){this.parserparser;}OverridepublicTdeserialize(Stringtopic,byte[]data){if(datanull){returnnull;}try{returnparser.parseFrom(data);// Protobuf自带反序列化}catch(InvalidProtocolBufferExceptione){thrownewSerializationException(Failed to deserialize,e);}}Overridepublicvoidclose(){// 无需释放资源}}4.3 使用示例// Producer端配置PropertiespropsnewProperties();props.put(key.serializer,org.apache.kafka.common.serialization.StringSerializer);props.put(value.serializer,com.example.ProtobufSerializer);// 使用自定义序列化器KafkaProducerString,OrderproducernewKafkaProducer(props);OrderorderOrder.newBuilder().setOrderId(1001L).setUserId(user_123).setAmount(99.99).setTimestamp(System.currentTimeMillis()).build();producer.send(newProducerRecord(orders,order.getUserId(),order));【Protobuf序列化 vs JSON序列化 对比】 Order对象 序列化结果 ┌─────────────────┐ ┌──────────────────────┐ │ orderId: 1001 │ Protobuf │ 08 E9 07 12 08 ... │ 约 25 bytes │ userId: user_123│ ──────────────► │ │ │ amount: 99.99 │ │ │ │ timestamp: ... │ JSON │ {orderId:1001, │ 约 120 bytes └─────────────────┘ ──────────────► │ userId:user_123,│ │ amount:99.99, │ │ timestamp:...} │ └──────────────────────┘ Protobuf序列化结果比JSON小约80%因为使用Varint变长编码五、实战Avro序列化器带Schema RegistryAvro比Protobuf更进一步——它把Schema模式定义和数据分离Schema可以独立管理数据本身不携带字段名更省空间。5.1 Avro序列化器实现importorg.apache.avro.Schema;importorg.apache.avro.generic.GenericRecord;importorg.apache.avro.io.*;importorg.apache.avro.specific.SpecificDatumWriter;importorg.apache.avro.specific.SpecificDatumReader;/** * Avro序列化器 * 使用Specific Record模式代码生成 */publicclassAvroSerializerTextendsSpecificRecordimplementsSerializerT{Overridepublicbyte[]serialize(Stringtopic,Tdata){if(datanull)returnnull;try{ByteArrayOutputStreamoutnewByteArrayOutputStream();BinaryEncoderencoderEncoderFactory.get().binaryEncoder(out,null);SpecificDatumWriterTwriternewSpecificDatumWriter(data.getSchema());// Avro二进制编码先写Schema指纹再写数据byte[]schemaFingerprintdata.getSchema().toString().getBytes();out.write(intToByteArray(schemaFingerprint.length));// 4字节长度out.write(schemaFingerprint);// Schema内容writer.write(data,encoder);// 数据内容encoder.flush();returnout.toByteArray();}catch(IOExceptione){thrownewSerializationException(Avro serialization failed,e);}}privatebyte[]intToByteArray(intvalue){returnnewbyte[]{(byte)(value24),(byte)(value16),(byte)(value8),(byte)value};}}/** * Avro反序列化器 */publicclassAvroDeserializerTextendsSpecificRecordimplementsDeserializerT{privatefinalClassTtargetType;publicAvroDeserializer(ClassTtargetType){this.targetTypetargetType;}OverridepublicTdeserialize(Stringtopic,byte[]data){if(datanull)returnnull;try{// 解析SchemaintschemaLenByteBuffer.wrap(data,0,4).getInt();StringschemaStrnewString(data,4,schemaLen);SchemaschemanewSchema.Parser().parse(schemaStr);// 解析数据ByteArrayInputStreaminnewByteArrayInputStream(data,4schemaLen,data.length-4-schemaLen);BinaryDecoderdecoderDecoderFactory.get().binaryDecoder(in,null);SpecificDatumReaderTreadernewSpecificDatumReader(schema);// 创建新实例注意Avro反序列化不回填原对象TresulttargetType.getDeclaredConstructor().newInstance();reader.read(result,decoder);returnresult;}catch(Exceptione){thrownewSerializationException(Avro deserialization failed,e);}}}5.2 Avro Schema定义{type:record,name:OrderAvro,namespace:com.example.avro,fields:[{name:orderId,type:long},{name:userId,type:string},{name:amount,type:double},{name:timestamp,type:long}]}六、序列化方案终极对比维度JSONProtobufAvro序列化大小大文本格式小二进制Varint最小Schema分离可读性✅ 人类可读❌ 二进制❌ 二进制Schema演化❌ 无约束✅ 向前兼容✅✅ 最佳独立Schema跨语言支持✅ 所有语言✅ 多语言✅ 多语言序列化速度慢文本解析快快需要Schema❌ 不需要编译时检查运行时检查适用场景调试开发、日志内部微服务数据管道、流处理与Kafka集成度一般好✅✅ 最佳Confluent原生七、生产环境最佳实践7.1 序列化器选择决策树┌─── 是否需要跨语言互操作 │ ┌──否──┤ ┌── 性能要求极高 │ │ │ │ └──是──► 消息量级 ──► 少量 ──► JSON简单可读 │ │ │ └── 海量 ──► Schema管理能力 │ │ └── JSON ┌─── 有 ──► Avro Schema Registry │ └─── 无 ──► Protobuf7.2 常见踩坑Key和Value序列化器必须成对配置Producer配了自定义SerializerConsumer必须配对应的Deserializernull值处理Kafka允许消息的Key或Value为null但自定义序列化器必须能处理null输入序列化异常传播serialize()中的异常要包装成SerializationException抛出不要复用序列化器状态如果serialize()里用了成员变量确保它是线程安全的——因为KafkaProducer是线程安全的一个Producer实例可能被多个线程共享本篇小结本文带你彻底搞懂了Kafka的序列化机制设计哲学Kafka只认识byte[]序列化/反序列化完全由客户端负责服务端什么都不管——这也是Kafka高性能的秘诀之一接口设计Serializer/Deserializer接口简单到只有三个方法开箱即用的内置实现覆盖了所有基本类型实战重点Protobuf序列化器适合内部微服务数据小、速度快Avro序列化器配合Schema Registry适合数据管道Schema演化能力最强选型建议调试用JSON内部微服务用Protobuf跨组织数据管道用Avro下一篇我们来聊分区器看看消息是怎么决定去哪个分区的上一篇【第12篇】Kafka拦截器源码解析——你不知道的消息预处理神器下一篇【第14篇】Kafka分区器源码解析——消息去哪个分区有学问
