我通过使用 Spring Boot 应用程序在 Kafka Producer 中将其转换为 toString() 来发送一个 JSON 数组,但我在 Consumer 中遇到以下错误:
org.springframework.kafka.listener.ListenerExecutionFailedException:无法使用传入消息调用侦听器方法 端点处理程序详细信息:方法 [public void com.springboot.service.KafkaReciever.recieveData(com.springboot.model.Student,java.lang.String) throws java.lang.Exception] Bean [com.springboot.service.KafkaReciever@5bb3d42d ]; 嵌套异常是 org.springframework.messaging.converter.MessageConversionException:无法处理消息;嵌套异常是 org.springframework.messaging.converter.MessageConversionException:无法从 [java.lang.String] 转换为 [com.springboot.model.Student] for GenericMessage [payload=[com.springboot.model.Student@5e40dc31, com] .springboot.model.Student@235e68b8], headers={kafka_offset=45, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, kafka_receivedTopic=myTopic-kafkasender}], failedMessage=GenericMessage.Student@3dcmodel=GenericMessage. com。
配置文件:
@Configuration
@EnableKafka
public class KafkaConsumerConfig {
@Value("${kafka.boot.server}")
private String kafkaServer;
@Value("${kafka.consumer.group.id}")
private String kafkaGroupId;
@Bean
public ConsumerFactory<String, String> consumerConfig() {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaGroupId);
props.put("message.assembler.buffer.capacity", 33554432);
props.put("max.tracked.messages.per.partition", 24);
props.put("exception.on.message.dropped", true);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); …Run Code Online (Sandbox Code Playgroud)