序列化
-
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> {
@Override
public void configure(Map<String, ?> configs, boolean isKey) {
}
@Override
public byte[] serialize(String s, User user) {
return JSON.toJSONBytes(user);
}
@Override
public void close() {
}
}
public class UserDeserializer implements Deserializer<User> {
@Override
public 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);
DelegatingByTypeSerializer
从 2.8 开始新增了DelegatingByTypeSerializer完美解决了上面问题用法如下，
@Bean
public 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));
}
完整配置
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;
}
@Bean
public 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));
}
@Bean
public 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);
}结果验证
消息被正常消费
总结
本文主要介绍了自定义序列化方式，以及为不同类型指定不同的序列化实现方式。
由于篇幅有限，文中只包含核心配置和重要代码，部分代码未贴出，可留言交流学习。
