如何将Kafka JSON消息映射到Java对象？Spring Boot场景求助

你的核心问题是Kafka消息中的字段名与Order类的属性名不匹配，导致无法自动映射。以下是具体的修复步骤：

1. 修正Order类的字段映射 #

在Order类中，使用Jackson的@JsonProperty注解，将Kafka消息中的字段名与类属性一一对应：

import com.fasterxml.jackson.annotation.JsonProperty;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;

@Document("Order")
public class Order {

    @Id
    @JsonProperty("Id")
    private int id;
    
    @JsonProperty("FualType")
    private String type;
    
    @JsonProperty("Capacity")
    private int capacity;

    // 构造方法、getter、setter保持不变
    public Order() {}

    public Order(int id, String type, int capacity) {
        this.id = id;
        this.type = type;
        this.capacity = capacity;
    }

    public int getId() { return id; }
    public void setId(int id) { this.id = id; }
    public String getType() { return type; }
    public void setType(String type) { this.type = type; }
    public int getCapacity() { return capacity; }
    public void setCapacity(int capacity) { this.capacity = capacity; }
}

2. 确认Kafka容器工厂配置正确 #

确保你的orderKafkaListenerFactory配置了JSON反序列化器，指定Order类作为目标类型：

import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;

@EnableKafka
public class KafkaConsumerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Order> orderKafkaListenerFactory(
            ConsumerFactory<String, Order> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, Order> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        return factory;
    }

    @Bean
    public ConsumerFactory<String, Order> consumerFactory() {
        JsonDeserializer<Order> deserializer = new JsonDeserializer<>(Order.class);
        deserializer.addTrustedPackages("your.package.name"); // 替换为Order类所在的包路径
        deserializer.setUseTypeHeaders(false); // 不需要类型头可关闭，避免反序列化报错

        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // 替换为你的Kafka地址
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "group_json");
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, deserializer);

        return new DefaultKafkaConsumerFactory<>(configProps, new StringDeserializer(), deserializer);
    }
}

3. 验证消息接收并添加MongoDB存储逻辑 #

修改后，监听器能正确映射消息到Order对象列表，可直接添加计算和存储逻辑：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.stereotype.Component;
import java.util.List;

@Component
public class OrderConsumer {

    @Autowired
    private MongoTemplate mongoTemplate; // 或注入OrderRepository

    @KafkaListener(topics = "neworder", groupId = "group_json",
            containerFactory = "orderKafkaListenerFactory")
    public void consumeJson(List<Order> orders) {
        System.out.println("Consumed mapped orders: " + orders);
        
        // 执行自定义计算逻辑
        orders.forEach(order -> {
            int calculatedCapacity = order.getCapacity() * 2; // 示例计算
            System.out.println("Calculated capacity for order " + order.getId() + ": " + calculatedCapacity);
        });
        
        // 批量存储到MongoDB
        mongoTemplate.insertAll(orders);
        // 若用Repository：orderRepository.saveAll(orders);
    }
}

关键注意事项 #

• 确保Kafka消息是标准JSON格式，比如[{"Id":11,"FualType":"Petrol 92","Capacity":33000}]，若消息格式是非标准键值对字符串，需先转换为标准JSON再进行反序列化。
• 若Kafka发送的是单个Order对象而非数组，可将监听器参数改为Order order，同时保持反序列化配置不变。

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