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> {

    @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);

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

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));
    }

消费者序列化可以在   @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;
    }

    @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);
}

结果验证

消息被正常消费

总结

        本文主要介绍了自定义序列化方式，以及为不同类型指定不同的序列化实现方式。

        由于篇幅有限，文中只包含核心配置和重要代码，部分代码未贴出，可留言交流学习。