kafka 序列化和反序列化自定义实现

Kafka 是一个分布式流处理平台，它允许你发布和订阅记录流。在 Kafka 中，消息的传输需要经过序列化和反序列化的过程。默认情况下，Kafka 使用 Java 序列化，但你可以自定义序列化和反序列化的实现。

下面是一个简单的例子，展示了如何使用 JSON 作为序列化和反序列化的实现。我们将使用 org.apache.kafka.common.serialization 包中的 Serializer 和 Deserializer 接口。

首先，添加依赖到你的项目中。如果你使用的是 Maven，可以在 pom.xml 文件中添加以下依赖：

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.12.5</version>
</dependency>

接下来，创建一个自定义的序列化器（JsonSerializer）和反序列化器（JsonDeserializer）：

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.serialization.Deserializer;

import java.nio.ByteBuffer;
import java.util.Map;

public class JsonSerializer<T> implements Serializer<T> {
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 配置序列化器，例如设置日期格式等
    }

    @Override
    public byte[] serialize(String topic, T data) {
        try {
            return objectMapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new RuntimeException("Error serializing message", e);
        }
    }

    @Override
    public void close() {
        // 关闭序列化器，释放资源
    }
}

public class JsonDeserializer<T> implements Deserializer<T> {
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 配置反序列化器，例如设置日期格式等
    }

    @Override
    public T deserialize(String topic, byte[] data) {
        try {
            return objectMapper.readValue(data, objectMapper.constructType(getClass()));
        } catch (Exception e) {
            throw new RuntimeException("Error deserializing message", e);
        }
    }

    @Override
    public void close() {
        // 关闭反序列化器，释放资源
    }
}

现在，你可以在 Kafka Producer 和 Consumer 的配置中使用自定义的序列化器和反序列化器。例如，在 Kafka Producer 中使用自定义的 JSON 序列化器：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class CustomKafkaProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "com.example.JsonSerializer");

        KafkaProducer<String, MyMessage> producer = new KafkaProducer<>(props);
        MyMessage message = new MyMessage("Hello", "World");
        producer.send(new ProducerRecord<>("my-topic", message));
        producer.close();
    }
}

在 Kafka Consumer 中使用自定义的 JSON 反序列化器：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class CustomKafkaConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "com.example.JsonDeserializer");

        KafkaConsumer<String, MyMessage> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        while (true) {
            ConsumerRecords<String, MyMessage> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, MyMessage> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
}

这样，你就可以在 Kafka 中使用自定义的 JSON 序列化和反序列化实现了。注意，这里的例子仅用于演示目的，实际应用中可能需要根据具体需求进行调整。