1. 概述

Kafka傳輸自定義的DTO對象時，不能像平時一樣使用StringSerializer和StringDeserializer。這種情況需要自己實現對應DTO的序列化器和反序列化器。假設現在有個 KafkaMsgDto 類，代碼如下：

@Data public class KafkaMsgDto {private String id;private ActionEnum action;public KafkaMsgDto(){}public KafkaMsgDto(String id, ActionEnum action){this.id = id;this.action = action;}public enum ActionEnum{SAVE,DELETE;} }

2. Serializer

public class KafkaMsgSerializer implements Serializer<KafkaMsgDto> {private String encoding = "UTF8";public KafkaMsgSerializer(){}@Overridepublic void configure(Map<String, ?> configs, boolean isKey) {String propertyName = isKey ? "key.serializer.encoding" : "value.serializer.encoding";Object encodingValue = configs.get(propertyName);if (encodingValue == null) {encodingValue = configs.get("serializer.encoding");}if (encodingValue instanceof String) {this.encoding = (String)encodingValue;}}@Overridepublic byte[] serialize(String s, KafkaMsgDto data) {try {if (data == null){return null;}return JSON.toJSONString(data).getBytes(this.encoding);} catch (UnsupportedEncodingException var4) {throw new SerializationException("Error when serializing string to byte[] due to unsupported encoding " + this.encoding);}} }

3. Deserializer

public class KafkaMsgDeserializer implements Deserializer<KafkaMsgDto> {private String encoding = "UTF8";public KafkaMsgDeserializer(){}public void configure(Map<String, ?> configs, boolean isKey) {String propertyName = isKey ? "key.deserializer.encoding" : "value.deserializer.encoding";Object encodingValue = configs.get(propertyName);if (encodingValue == null) {encodingValue = configs.get("deserializer.encoding");}if (encodingValue instanceof String) {this.encoding = (String)encodingValue;}}public KafkaMsgDto deserialize(String topic, byte[] data) {try {if (data == null){return null;}return JSON.parseObject(new String(data, this.encoding), KafkaMsgDto.class);} catch (UnsupportedEncodingException var4) {throw new SerializationException("Error when deserializing byte[] to string due to unsupported encoding " + this.encoding);}} }

4. application.properties

修改對應的序列化、反序列化配置

spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=com.train.kafka.serialization.KafkaMsgSerializerspring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=com.train.kafka.serialization.KafkaMsgDeserializer

5. 發送消息

KafkaMsgDto dto = new KafkaMsgDto(id, KafkaMsgDto.ActionEnum.SAVE); kafkaTemplate.send("test-topic", dto);

6. 接收消息

@KafkaListener(topics = "test-topic")public void onListener(KafkaMsgDto dto){if (dto == null){System.out.println("接收到空消息");return;}System.out.println(String.format("接收到消息：%s", dto.toString()));if (dto.getAction() == KafkaMsgDto.ActionEnum.DELETE){//執行刪除業務邏輯}else{//執行保存業務邏輯}}

總結

以上是生活随笔為你收集整理的SpringBoot 自定义Kafka消息序列化和反序列化的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。