kafkabinder自定义序列化器

在使用Spring Cloud Stream和Kafka Binder时，可以通过自定义序列化器来实现消息的序列化和反序列化。具体步骤如下：

1. 创建自定义的序列化器类，实现Spring的Serializer和Deserializer接口。例如，创建一个Person类和序列化器。

public class Person {
    private String name;
    private int age;
    // ...
}

public class PersonSerializer implements Serializer<Person>, Deserializer<Person> {
    @Override
    public byte[] serialize(String topic, Person data) {
        // 将Person对象序列化为字节数组
        ObjectMapper mapper = new ObjectMapper();
        try {
            return mapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public Person deserialize(String topic, byte[] data) {
        // 将字节数组反序列化为Person对象
        ObjectMapper mapper = new ObjectMapper();
        try {
            return mapper.readValue(data, Person.class);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

2. 在application.yml中配置自定义序列化器。需要设置属性spring.cloud.stream.bindings.<channelName>.producer.serialization[Type|Class]和spring.cloud.stream.bindings.<channelName>.consumer.serialization[Type|Class]，分别表示生产者和消费者的序列化器类型或类名。例如：

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
      bindings:
        input:
          destination: testTopic
          group: testGroup
          consumer:
            serializationClass: com.example.PersonSerializer
          producer:
            serializationClass: com.example.PersonSerializer
        output:
          destination: testTopic
          group: testGroup
          consumer:
            serializationClass: com.example.PersonSerializer
          producer:
            serializationClass: com.example.PersonSerializer

3. 在代码中使用自定义序列化器。例如，使用Spring的StreamListener注解，将消息转换为Person对象。

@Service
@EnableBinding(Sink.class)
public class KafkaConsumer {
    @StreamListener(Sink.INPUT)
    public void processMessage(Message<Person> message) {
        Person person = message.getPayload();
        // 处理逻辑
    }
}