kafka 配置自定义序列化方式

序列化

kafka 需要将消息内容序列化（Serializer）成字节数组才能发送到 Broken节点
消费者需要将字节数组反序列化（Deserializer）为消息内容，然后消费消息。接口定义如下

public interface Serializer<T> extends Closeable {default void configure(Map<String, ?> configs, boolean isKey) {}byte[] serialize(String var1, T var2);default byte[] serialize(String topic, Headers headers, T data) {return this.serialize(topic, data);}default void close() {}
}

public interface Deserializer<T> extends Closeable {default void configure(Map<String, ?> configs, boolean isKey) {}T deserialize(String var1, byte[] var2);default T deserialize(String topic, Headers headers, byte[] data) {return this.deserialize(topic, data);}default void close() {}
}

常用序列化器

kafka 内置了许多实现如 StringSerializer、IntegerSerializer、DoubleSerializer 等。

生产者与消费者选择的序列化与反序列化要匹配才能正常解析

自定义序列化器

自定义序列化器只需要实现对应的 Serializer,UserDeserializer

比如有一个 User 类

@Data
@NoArgsConstructor
@AllArgsConstructor
public class User {private Long id;private String userName;
}

UserSerializer

public class UserSerializer implements Serializer<User> {@Overridepublic void configure(Map<String, ?> configs, boolean isKey) {}@Overridepublic byte[] serialize(String s, User user) {return JSON.toJSONBytes(user);}@Overridepublic void close() {}
}

public class UserDeserializer implements Deserializer<User> {@Overridepublic User deserialize(String s, byte[] bytes) {return JSON.parseObject(bytes,User.class);}
}

生产者与消费者配置

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, UserDeserializer.class);
...
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, UserSerializer.class);

上面我们配置了 User 相关的序列化，但是实际上项目发送的消息内容肯定不止 User 这一种，我们需要针对不同类型指定不同的序列化方式

DelegatingByTypeSerializer

从 2.8 开始新增了DelegatingByTypeSerializer完美解决了上面问题用法如下，

    @Beanpublic ProducerFactory producerFactory() {Map<Class<?>, Serializer> delegates = new HashMap<>();delegates.put(byte[].class, new ByteArraySerializer());delegates.put(Bytes.class, new BytesSerializer());delegates.put(String.class, new StringSerializer());delegates.put(User.class, new UserSerializer());return new DefaultKafkaProducerFactory<>(produceConfigs(),new StringSerializer(), new DelegatingByTypeSerializer(delegates));}

消费者序列化可以在 @KafkaListener 指定

完整配置

确保kafka-clients版本在2.8以上

kafka 配置类

@Configuration
@EnableKafka
public class KafkaConfig {private Map<String, Object> produceConfigs() {Map<String, Object> configMap = new HashMap<>();configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"ip:9092");configMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);return configMap;}@Beanpublic ProducerFactory producerFactory() {Map<Class<?>, Serializer> delegates = new HashMap<>();delegates.put(byte[].class, new ByteArraySerializer());delegates.put(Bytes.class, new BytesSerializer());delegates.put(String.class, new StringSerializer());delegates.put(User.class, new UserSerializer());return new DefaultKafkaProducerFactory<>(produceConfigs(),new StringSerializer(), new DelegatingByTypeSerializer(delegates));}@Beanpublic KafkaTemplate kafkaTemplate() {return new KafkaTemplate(producerFactory());}private Map<String, Object> consumerConfigs() {Map<String, Object> configMap = new HashMap<>();configMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"ip:9092");configMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,IntegerDeserializer.class);configMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);configMap.put(ConsumerConfig.GROUP_ID_CONFIG, "groupTest");return configMap;}}

消费者 @KafkaListener 注解可以指定序列化器

 @KafkaListener(topics = "test")public void processMessage(final ConsumerRecord<String,User> record) {System.out.println("processMessage:" + JSON.toJSONString(record.value()));}@KafkaListener(topics = "topic2",properties = "value-deserializer:org.apache.kafka.common.serialization.StringDeserializer")public void processMessage(String content) {System.out.println("processMessage:" + content);}

发送消息代码

for (int i = 0; i < 100; i++) {messageProduce.sendMessage(new User(Long.valueOf(i), "zhangsan" + i));messageProduce.sendMessage2("hello world" + i);
}