springboot kafka如何進行消息轉換

在Spring Boot中，我們可以使用Kafka的ConsumerMessageConverter和ProducerMessageConverter進行消息轉換。這兩個類分別用于將Kafka消費者和生產者接收到的消息進行序列化和反序列化。為了實現自定義的消息轉換，我們可以創建一個實現org.springframework.kafka.support.converter.MessageConverter接口的類，并重寫convertFromInternal和convertToInternal方法。

以下是一個簡單的示例，展示了如何創建一個自定義的消息轉換器并將其應用于Spring Boot Kafka配置：

1. 首先，創建一個實現MessageConverter接口的類：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.MessageConversionException;
import org.springframework.kafka.support.serializer.RecordMetadata;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.handler.annotation.Header;

import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;

public class CustomKafkaMessageConverter extends MappingJackson2MessageConverter {

    public CustomKafkaMessageConverter() {
        super();
        setCharset(Charset.forName("UTF-8"));
        addPayloadDeserializer(new JsonDeserializer<>());
        addHeaderDeserializer(new StringDeserializer());
    }

    @Override
    protected Object convertFromInternal(Object payload, MessageHeaders headers, byte[] bytes) throws MessageConversionException {
        // 在這里實現自定義的反序列化邏輯
        return super.convertFromInternal(payload, headers, bytes);
    }

    @Override
    protected byte[] convertToInternal(Object payload, MessageHeaders headers) throws MessageConversionException {
        // 在這里實現自定義的序列化邏輯
        return super.convertToInternal(payload, headers);
    }
}

2. 然后，在Spring Boot配置類中，將自定義的消息轉換器應用于Kafka消費者和生產者：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListenerConfigurer;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpointRegistrar;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.config.MethodKafkaListenerEndpoint;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaConfig implements KafkaListenerConfigurer {

    @Bean
    public CustomKafkaMessageConverter customKafkaMessageConverter() {
        return new CustomKafkaMessageConverter();
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory,
            ProducerFactory<String, String> producerFactory) {

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setProducerFactory(producerFactory);
        factory.setMessageConverter(customKafkaMessageConverter());
        return factory;
    }

    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);

        ConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(props);

        Map<String, Object> producerProps = new HashMap<>();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(producerProps);

        registrar.registerEndpoint(new MethodKafkaListenerEndpoint<>(
                "testEndpoint",
                "testMethod",
                new StringDeserializer(),
                new StringDeserializer(),
                kafkaListenerContainerFactory(consumerFactory, producerFactory)
        ));
    }

    @Override
    public void configureKafkaProducers(KafkaProducerFactory<String, String> factory) {
        // 配置生產者屬性，如果需要的話
    }
}

現在，當使用@KafkaListener注解監聽Kafka主題時，消息將使用自定義的消息轉換器進行序列化和反序列化。