Spring Boot使用Kafka发送JSON对象报类型转换错误求助

问题现象 #

发送BookES对象到Kafka时抛出类型转换异常，提示无法将BookES实例转换为String，Postman请求返回500内部错误：

java.lang.ClassCastException: class com.example.demo.model.BookES cannot be cast to class java.lang.String (com.example.demo.model.BookES is in unnamed module of loader 'app'; java.lang.String is in module java.base of loader 'bootstrap')

错误响应详情：

{
    "timestamp": "2022-11-22T11:24:30.738+00:00",
    "status": 500,
    "error": "Internal Server Error",
    "message": "Can't convert value of class com.example.demo.model.BookES to class org.apache.kafka.common.serialization.StringSerializer specified in value.serializer",
    "path": "/books/publish"
}

核心原因 #

尽管配置了JSON序列化器，但Kafka生产者实际仍在使用StringSerializer处理BookES对象，说明序列化器配置未生效或配置错误。

解决方案 #

1. 修正application.properties配置 #

确保生产者的value.serializer指定为Spring Kafka提供的JSON序列化器，而非字符串序列化器：

# Kafka基础配置
spring.kafka.producer.bootstrap-servers=你的Kafka集群地址
# 键序列化器（若使用String作为键则保留）
spring.kafka.producer.key.serializer=org.apache.kafka.common.serialization.StringSerializer
# 关键：配置JSON值序列化器
spring.kafka.producer.value.serializer=org.springframework.kafka.support.serializer.JsonSerializer

2. 确保KafkaTemplate使用正确的生产者配置 #

如果自定义了生产者工厂，需显式指定值序列化器类型：

@Configuration
public class KafkaProducerConfig {
    @Value("${spring.kafka.producer.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public ProducerFactory<String, BookES> bookProducerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 指定JsonSerializer处理BookES对象
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, BookES> bookKafkaTemplate() {
        return new KafkaTemplate<>(bookProducerFactory());
    }
}

3. 规范BookES模型类 #

无需添加@JsonSerializer注解，只需保证类是标准POJO：

• 包含无参构造函数（Jackson序列化必须）
• 字段提供getter/setter方法

public class BookES {
    private Long id;
    private String title;
    private String author;

    // 无参构造函数
    public BookES() {}

    // 带参构造函数（可选）
    public BookES(Long id, String title, String author) {
        this.id = id;
        this.title = title;
        this.author = author;
    }

    // getter和setter方法
    public Long getId() { return id; }
    public void setId(Long id) { this.id = id; }
    public String getTitle() { return title; }
    public void setTitle(String title) { this.title = title; }
    public String getAuthor() { return author; }
    public void setAuthor(String author) { this.author = author; }
}

4. 检查生产者发送逻辑 #

确保直接发送BookES对象，而非手动转换为字符串：

@Service
public class KafkaBookProducer {
    private final KafkaTemplate<String, BookES> bookKafkaTemplate;

    @Autowired
    public KafkaBookProducer(KafkaTemplate<String, BookES> bookKafkaTemplate) {
        this.bookKafkaTemplate = bookKafkaTemplate;
    }

    public void publishBook(BookES book) {
        // 直接发送BookES实例
        bookKafkaTemplate.send("book-topic", book);
    }
}

5. 确认依赖完整性 #

确保项目引入Spring Kafka及Jackson依赖（Maven示例）：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>适配你的Spring Boot版本</version>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

内容的提问来源于stack exchange，提问作者chinmay