spring整合kafka如何进行消息转换

在Spring整合Kafka时，消息转换是一个重要的环节。为了实现消息转换，你可以使用Kafka的MessageConverter接口。Spring提供了两种常用的MessageConverter实现：StringHttpMessageConverter和ByteArrayHttpMessageConverter。但在Kafka消费者端，我们通常使用KafkaDeserializer来处理消息的反序列化。

以下是一个简单的示例，展示了如何在Spring整合Kafka时进行消息转换：

首先，创建一个自定义的KafkaDeserializer，用于实现消息的反序列化。在这个例子中，我们将把字符串消息转换为MyMessage对象：

import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.databind.ObjectMapper; import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer; import org.springframework.kafka.support.serializer.JsonDeserializer; import java.io.IOException; public class MyKafkaDeserializer extends ErrorHandlingDeserializer<MyMessage> { private final ObjectMapper objectMapper = new ObjectMapper(); public MyKafkaDeserializer() { this(null); } public MyKafkaDeserializer(String keyDeserializer) { super(new JsonDeserializer<>(MyMessage.class)); } @Override protected MyMessage deserialize(String topic, byte[] data) throws IOException { return objectMapper.readValue(data, MyMessage.class); } }

在你的Kafka消费者配置类中，将自定义的MyKafkaDeserializer添加到KafkaListenerEndpointRegistry中：

import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.serialization.StringDeserializer; import org.springframework.beans.factory.annotation.Value; 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.listener.ConcurrentMessageListenerContainer; import org.springframework.kafka.listener.config.MethodKafkaListenerEndpoint; import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer; import org.springframework.kafka.support.serializer.JsonDeserializer; import java.util.HashMap; import java.util.Map; @Configuration public class KafkaConsumerConfig implements KafkaListenerConfigurer { @Value("${spring.kafka.bootstrap-servers}") private String bootstrapServers; @Bean public ConcurrentKafkaListenerContainerFactory<String, MyMessage> kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory<String, MyMessage> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); return factory; } @Bean public ConsumerFactory<String, MyMessage> consumerFactory() { return new DefaultKafkaConsumerFactory<>(consumerConfigs()); } @Bean public Map<String, Object> consumerConfigs() { Map<String, Object> props = new HashMap<>(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group"); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, MyKafkaDeserializer.class); return props; } @Override public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) { MethodKafkaListenerEndpoint<String, MyMessage> endpoint = new MethodKafkaListenerEndpoint<>(); endpoint.setId("myListener"); endpoint.setTopics("my-topic"); endpoint.setMessageHandlerMethodFactory(kafkaListenerContainerFactory().getMessageHandlerMethodFactory()); endpoint.setBean(new MyKafkaConsumer()); registrar.registerEndpoint(endpoint); } @Override public void configureKafkaListenerEndpointRegistrars(KafkaListenerEndpointRegistry registry) { registry.registerEndpoints(configureKafkaListeners(new KafkaListenerEndpointRegistrar())); } }

创建一个Kafka消费者监听器类MyKafkaConsumer，用于处理接收到的消息：

import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component; @Component public class MyKafkaConsumer { @KafkaListener(id = "myListener", groupId = "my-group") public void listen(MyMessage message) { System.out.println("Received message: " + message); } }

最后，确保你的MyMessage类已经正确实现了序列化和反序列化。你可以使用Jackson库来实现这个功能：

import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; public class MyMessage { private String content; // Getters and setters public static MyMessage fromJson(String json) throws JsonProcessingException { ObjectMapper objectMapper = new ObjectMapper(); return objectMapper.readValue(json, MyMessage.class); } public String toJson() throws JsonProcessingException { ObjectMapper objectMapper = new ObjectMapper(); return objectMapper.writeValueAsString(this); } }

现在，当你的Spring应用从Kafka消费消息时，消息将被转换为MyMessage对象，然后由MyKafkaConsumer处理。

最新问答

相关标签