动态创建Spring Cloud Stream消费者丢失消息类型，无法获取消息头

问题场景 #

• 用@Bean注解定义的Consumer<Message<SitePayload>>能正常接收包含消息头和负载的完整Message对象；
• 通过BeanFactoryPostProcessor调用registerSingleton动态注册的消费者，收到的输入是byte[]（序列化后的SitePayload），而且拿不到消息头——原因是registerSingleton接收Object类型参数，泛型类型信息被擦除，Spring Cloud Stream无法识别该消费者需要接收Message类型，只能直接传递序列化后的负载内容。

解决方案 #

核心是让Spring保留消费者的泛型类型信息，不能直接用registerSingleton（会擦除泛型），而是通过GenericBeanDefinition明确指定Bean的泛型类型，再注册到BeanFactory中。

修改后的代码示例：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.GenericBeanDefinition;
import org.springframework.cloud.stream.function.StreamFunctionUtils;
import org.springframework.messaging.Message;
import java.util.function.Consumer;

public class ConsumersCreator implements BeanFactoryPostProcessor {

    private static final Logger log = LoggerFactory.getLogger(ConsumersCreator.class);

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
        BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory;

        for (int i = 0; i < 5; i++) {
            String beanName = "consumer" + i;
            // 创建GenericBeanDefinition来保留泛型类型信息
            GenericBeanDefinition beanDefinition = new GenericBeanDefinition();
            beanDefinition.setBeanClass(Consumer.class);
            // 明确指定泛型参数：Consumer<Message<SitePayload>>
            StreamFunctionUtils.configureFunctionType(beanDefinition, Consumer.class, Message.class, SitePayload.class);
            // 先注册BeanDefinition到容器
            registry.registerBeanDefinition(beanName, beanDefinition);
            // 再注册具体的消费者实例
            beanFactory.registerSingleton(beanName, createConsumer(i));
        }
    }

    private Consumer<Message<SitePayload>> createConsumer(int id) {
        return message -> log.info("Consumer id: {} | 负载内容: {} | 消息头: {}", 
            id, message.getPayload(), message.getHeaders());
    }

    @Bean
    public Consumer<Message<SitePayload>> consumer() {
        return message -> log.error("固定消费者 | 负载内容: {} | 消息头: {}", 
            message.getPayload(), message.getHeaders());
    }
}

关键说明 #

• StreamFunctionUtils.configureFunctionType是Spring Cloud Stream提供的工具方法，用来给函数Bean设置泛型类型信息，让框架能识别消费者需要接收的是Message类型，而非直接处理负载；
• 先注册GenericBeanDefinition保留类型元数据，再注册Singleton实例，这样Spring容器就能正确识别消费者的泛型类型，传递完整的Message对象（包含消息头和负载）。

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