如何在Spring Integration中基于配置动态创建多个JMS消息驱动通道适配器？

当然可以做到！在Spring 4.3.4 + Java 8的环境下，Spring Integration完全支持从配置读取队列详情，自动生成对应的消费者Bean并注册到应用上下文，不用手动编写几十份重复的Bean定义。下面是具体的实现思路和示例：

核心思路 #

利用Spring的BeanDefinitionRegistryPostProcessor接口——这个接口允许我们在Spring上下文初始化的Bean定义阶段，动态注册新的Bean定义。结合Spring Integration的API，我们可以遍历配置中的队列列表，为每个队列生成对应的消费者组件（比如JMS/RabbitMQ的消息驱动适配器），并将它们注册到上下文。

具体实现步骤 #

1. 配置队列列表 #

首先在你的配置文件（比如application.properties）中定义需要监听的队列列表：

# 逗号分隔的队列名称
app.jms.queues=order-processing,payment-notify,inventory-update,user-signup,...
# 可选：统一配置消费者参数
app.jms.consumer.concurrent-consumers=2
app.jms.consumer.max-concurrent-consumers=5

2. 编写动态Bean注册处理器 #

创建一个实现BeanDefinitionRegistryPostProcessor和EnvironmentAware的类，用来读取配置并注册消费者Bean：

import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.context.EnvironmentAware;
import org.springframework.core.env.Environment;
import org.springframework.integration.jms.JmsMessageDrivenChannelAdapter;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.beans.factory.support.RuntimeBeanReference;

public class DynamicJmsConsumerRegistrar implements BeanDefinitionRegistryPostProcessor, EnvironmentAware {

    private Environment environment;

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = environment;
    }

    @Override
    public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {
        // 读取配置中的队列列表
        String queueConfig = environment.getProperty("app.jms.queues");
        if (queueConfig == null || queueConfig.trim().isEmpty()) {
            return;
        }

        String[] queueNames = queueConfig.split(",");
        int concurrentConsumers = environment.getProperty("app.jms.consumer.concurrent-consumers", Integer.class, 2);
        int maxConcurrentConsumers = environment.getProperty("app.jms.consumer.max-concurrent-consumers", Integer.class, 5);

        for (String queue : queueNames) {
            String cleanQueueName = queue.trim();
            // 1. 注册消息监听容器（消费者的核心运行容器）
            RootBeanDefinition containerDef = new RootBeanDefinition(DefaultMessageListenerContainer.class);
            containerDef.getPropertyValues().add("connectionFactory", new RuntimeBeanReference("jmsConnectionFactory"));
            containerDef.getPropertyValues().add("destinationName", cleanQueueName);
            containerDef.getPropertyValues().add("concurrentConsumers", concurrentConsumers);
            containerDef.getPropertyValues().add("maxConcurrentConsumers", maxConcurrentConsumers);

            String containerBeanName = cleanQueueName + "-listener-container";
            registry.registerBeanDefinition(containerBeanName, containerDef);

            // 2. 注册Spring Integration的消息驱动适配器（连接容器和消息通道）
            RootBeanDefinition adapterDef = new RootBeanDefinition(JmsMessageDrivenChannelAdapter.class);
            adapterDef.getPropertyValues().add("messageListenerContainer", new RuntimeBeanReference(containerBeanName));
            adapterDef.getPropertyValues().add("outputChannel", new RuntimeBeanReference("commonProcessingChannel"));

            String adapterBeanName = cleanQueueName + "-consumer-adapter";
            registry.registerBeanDefinition(adapterBeanName, adapterDef);
        }
    }

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
        // 无需额外处理
    }
}

3. 注册处理器到Spring上下文 #

通过Java配置或者XML配置，将上面的注册器Bean添加到上下文：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.MessageChannel;
import org.springframework.jms.connection.SingleConnectionFactory;
import org.springframework.jms.core.JmsTemplate;

@Configuration
public class IntegrationConfig {

    // 注册动态消费者注册器
    @Bean
    public DynamicJmsConsumerRegistrar dynamicJmsConsumerRegistrar() {
        return new DynamicJmsConsumerRegistrar();
    }

    // 公共消息处理通道（所有消费者的输出通道）
    @Bean
    public MessageChannel commonProcessingChannel() {
        return new DirectChannel();
    }

    // JMS连接工厂示例（根据你的MQ供应商调整）
    @Bean
    public SingleConnectionFactory jmsConnectionFactory() {
        SingleConnectionFactory factory = new SingleConnectionFactory();
        factory.setTargetConnectionFactory(new org.apache.activemq.ActiveMQConnectionFactory("tcp://localhost:61616"));
        return factory;
    }
}

适配其他队列类型 #

如果你的队列不是JMS（比如RabbitMQ的队列），只需要替换对应的消费者组件即可：

• 对于RabbitMQ，使用AmqpInboundChannelAdapter作为适配器，SimpleMessageListenerContainer作为监听容器，思路完全一致。

注意事项 #

• 确保配置中的队列名称和对应的连接工厂等依赖Bean已经正确配置，否则动态注册的Bean会初始化失败。
• 可以针对每个队列单独配置参数（比如并发数），只需要修改配置读取逻辑，比如用app.jms.queues.${queueName}.concurrent-consumers这样的配置键。
• Spring 4.3.4完全兼容BeanDefinitionRegistryPostProcessor接口，这个接口从Spring 3.0就存在，版本上没有问题。

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