发送到Kafka主题时反序列化对象时出错

Question

我对Kafka很陌生。我正在尝试发送一个消息到Kafka主题，其中包含头和有效载荷。

以下是错误：

"org.apache.kafka.common.errors.SerializationException: Can't convert value of class com.cabservice.request.CabLocationPayload to class org.apache.kafka.common.serialization.StringSerializer specified in value.serializer\nCaused by: java.lang.ClassCastException: class com.cabservice.request.CabLocationPayload cannot be cast to class java.lang.String 

public class CabLocationPayload {

private Header header;

private Payload payload;

@PostMapping(value=“/publish”）public void sendMessageToKafkaTopic(@RequestBody CabLocationPayload CabLocationPayload）{

Header和Payload具有JSON的映射字段。

在Producer中更改VALUE_SERIALIZER_CLASS_CONFIG后，我可以看到数据。但ClassCastException仍然失败。

    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

    return new DefaultKafkaProducerFactory<>(configProps);
}

@Bean
public KafkaTemplate<String, String> kafkaTemplateString() {
    return new KafkaTemplate<>(producerFactoryString());
}

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "group_id");
    configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);

    return new DefaultKafkaConsumerFactory<>(configProps);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());

    return factory;
}

当前错误为{2 09:41:20.108信息22561--[ad producer-1]org.apache.kafka.clients.metadata:[ProducerClientID=producer-1]群集ID：lWghv-b_RG-_hO-qOp_cjA 2021-04-22 09:41:20.123错误22561--[nio-9080-exec-2]o.a.cc.[.[.[/].[dispatcherServlet]:路径为[]的上下文中servlet[dispatcherServlet]的servlet.Service（）引发异常[请求处理失败；嵌套异常为

java.lang.ClassCastException：类com.cabservice.request.cabLocationPayload不能强制转换为类java.lang.String（com.cabservice.request.cabLocationPayload位于加载程序org.springframework.boot.devtools.restart.ClassLoader.restartClassLoader@1144043d的未命名模块中；java.lang.String位于加载程序“bootstrap”的模块java.base中），位于

任何帮助都是非常感谢的。

Answer 1

Kafka配置“value.Serializer”配置应该是序列化程序子类，而不是您的对象类型

例如：

键:VALUE_SERIALIZER_CLASS_CONFIG，值:jsonserializer.class（来源:org.springframework.kafka.support.serializer)

    @EnableKafka
    @Configuration
    public class KafkaProducerConfiguration {
    
        @Bean
        KafkaTemplate<String, Object> kafkaTemplate() {
            return new KafkaTemplate<>(producerFactory());
        }
    
        @Bean
        public ProducerFactory<String, Object> producerFactory() {
            return new DefaultKafkaProducerFactory<>(getConfig());
        }
    
        private Map<String, Object> getConfig() {
            Map<String, Object> config = new HashMap<>();
    
            config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "brokers");
            config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
            return config;
        }
    
    }

您必须将您的类替换为您想要消耗的类名。（对于本例：CabLocationPayload）

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
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 KafkaConsumerConfiguration {


    private Map<String, Object> consumerConfigs() {
        final Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your brokers");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumeer-group-id");
        return props;
    }

    @Bean
    public ConsumerFactory<String, YourClass> kafkaListenerConsumerFactory() {
        final ErrorHandlingDeserializer<YourClass> errorHandlingDeserializer = new ErrorHandlingDeserializer<>(new JsonDeserializer<>(YourClass.class, false));
        return new DefaultKafkaConsumerFactory<>(this.consumerConfigs(), new StringDeserializer(), errorHandlingDeserializer);
    }


    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, YourClass> kafkaListenerContainerFactory() {
        final ConcurrentKafkaListenerContainerFactory<String, YourClass> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(this.kafkaListenerConsumerFactory());
        return factory;
    }

}