Spring Kafka如何配置通用Consumer Factory以支持多模型类消息消费？

嗨，很高兴看到你已经成功搞定了单个User模型的Kafka消息收发！针对你提到的20+模型类不想逐个写ConsumerFactory的问题，完全不用这么麻烦，这里有两种非常实用的通用配置方案：

方案一：使用泛型消息包装类 #

我们可以定义一个通用的消息包装对象，把所有业务模型类都包裹在这个对象里，消费者只需要处理这个包装类，再根据类型字段反序列化到具体的业务模型。

步骤1：定义泛型包装类 #

public class GenericKafkaMessage<T> {
    private String payloadType; // 标记实际模型类的全限定名，比如com.example.User
    private T payload;

    // getters、setters、构造器
}

步骤2：生产者发送消息 #

发送时把业务对象包装成GenericKafkaMessage，并指定payloadType：

@Autowired
private KafkaTemplate<String, GenericKafkaMessage<?>> kafkaTemplate;

public void sendUserMessage(User user) {
    GenericKafkaMessage<User> message = new GenericKafkaMessage<>();
    message.setPayloadType(User.class.getCanonicalName());
    message.setPayload(user);
    kafkaTemplate.send("user-topic", message);
}

// 发送Product同理
public void sendProductMessage(Product product) {
    GenericKafkaMessage<Product> message = new GenericKafkaMessage<>();
    message.setPayloadType(Product.class.getCanonicalName());
    message.setPayload(product);
    kafkaTemplate.send("product-topic", message);
}

步骤3：通用消费者配置 #

只需要创建一个针对GenericKafkaMessage的ConsumerFactory和容器工厂：

@Configuration
public class GenericConsumerConfig {
    @Bean
    public ConsumerFactory<String, GenericKafkaMessage<?>> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG,"group1");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        // 无需指定具体业务类，后续自行处理类型转换
        return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(), new JsonDeserializer<>(GenericKafkaMessage.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, GenericKafkaMessage<?>> genericKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, GenericKafkaMessage<?>> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

步骤4：消费者接收并转换 #

在@KafkaListener里根据payloadType反射转换成具体模型：

@KafkaListener(topics = {"user-topic", "product-topic"}, containerFactory = "genericKafkaListenerContainerFactory")
public void handleMessage(GenericKafkaMessage<?> message) throws ClassNotFoundException {
    Object payload = message.getPayload();
    // 根据类型做不同业务处理
    if (payload instanceof User) {
        handleUser((User) payload);
    } else if (payload instanceof Product) {
        handleProduct((Product) payload);
    }
    // 其他模型类同理
}

private void handleUser(User user) {
    // User业务逻辑
}

private void handleProduct(Product product) {
    // Product业务逻辑
}

方案二：利用Spring Kafka JsonDeserializer的类型头自动推断 #

Spring Kafka的JsonDeserializer支持通过Kafka消息头传递目标类型信息，不需要在构造器里指定具体类，就能自动反序列化到对应的模型类。

步骤1：修改消费者配置 #

去掉JsonDeserializer构造器里的具体类，添加类型头相关配置：

@Configuration
public class GenericConsumerConfig {
    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG,"group1");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        // 开启类型头推断
        config.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, true);
        // 指定允许反序列化的类包名，避免安全风险
        config.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example.model");
        return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(), new JsonDeserializer<>());
    }

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

步骤2：生产者发送消息（自动携带类型头） #

使用KafkaTemplate发送时，只要指定泛型类型，Spring Kafka会自动在消息头里添加类型信息：

@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;

public void sendUserMessage(User user) {
    kafkaTemplate.send("user-topic", user);
}

public void sendProductMessage(Product product) {
    kafkaTemplate.send("product-topic", product);
}

步骤3：消费者接收消息 #

在@KafkaListener里直接指定接收的类型，或者用Object接收后判断：

// 针对单个主题指定类型
@KafkaListener(topics = "user-topic", containerFactory = "kafkaListenerContainerFactory")
public void handleUser(User user) {
    // User业务逻辑
}

@KafkaListener(topics = "product-topic", containerFactory = "kafkaListenerContainerFactory")
public void handleProduct(Product product) {
    // Product业务逻辑
}

// 或者多个主题用Object接收后判断类型
@KafkaListener(topics = {"user-topic", "product-topic"}, containerFactory = "kafkaListenerContainerFactory")
public void handleAllMessages(Object payload) {
    if (payload instanceof User) {
        handleUser((User) payload);
    } else if (payload instanceof Product) {
        handleProduct((Product) payload);
    }
}

小提示 #

顺便提一句，你原来的ConsumerFactory代码里，key用了StringSerializer，但消费者应该用StringDeserializer哦，虽然你说运行正常，但这大概率是复制时的小失误，记得修正~

两种方案各有优劣：

• 方案一的包装类更灵活，适合需要统一处理消息元数据（比如类型、时间戳）的场景；
• 方案二更简洁，利用Spring Kafka原生特性，代码量更少，适合大多数普通业务场景。

你可以根据自己的业务需求选择最合适的方案！

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