使用 Spring Integration 进行动态配置答案

【问题标题】：Dynamic configuration with Spring Integration使用 Spring Integration 进行动态配置
【发布时间】：2015-12-07 05:07:12
【问题描述】：

从 RabbitMQ 连接规范列表到发布订阅通道上的一系列消费者的正确方法是什么？

也就是说，我的兔子配置如下：

rabbits:
    - hostname: blahblah
      vhost: blahlbah
      username: blahlbah
      password: blahlbalh
      exchange: blahblah
    - hostname: blahblah1
      vhost: blahlbah1
      username: blahlbah1
      password: blahlbalh1
      exchange: blahblah1
    ...

我将这些编组为@NestedConfigurationProperty List<RabbitConfiguration>。我可以编写一个@Bean 方法，从其中一个List<RabbitConfiguration> 中获取AmqpTemplate，如下所示：

@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public AmqpTemplate amqpTemplate(RabbitConfiguration rabbitConfiguration) {
    ...
}

然后我基本上可以将其映射到IntegrationFlow：

@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public IntegrationFlow integrationFlow(AmqpTemplate amqpTemplate) {
    return IntegrationFlows.from(inboundPubSubChannel()).handle(Amqp.outboundAdapter(amqpTemplate)).get();
}

但是，对于 List<RabbitConfiguration> 执行此操作并让 Spring 处理生成的 List<IntegrationFlow> 的正确方法是什么？当然不只是：

@Bean
public List<IntegrationFlow> integrationFlows(List<RabbitConfiguration> rabbitConfigurations) {
    return rabbitConfigurations.stream()
        .map(this::amqpTemplate)
        .map(this::integrationFlow)
        .collect(toList())
}

【问题讨论】：

标签： java spring rabbitmq spring-integration spring-java-config

【解决方案1】：

我认为您要做的是根据您的配置属性动态创建 IntegrationFlow 类型的 Spring bean。您有很多选择，具体取决于您希望它有多“神奇”或“透明”。如果你想要完整的魔法，你必须实现一个 BeanFactoryPostProcessor 并在 Spring 上下文中注册。像这样的东西应该可以工作：

    public class IntegrationFlowPostProcessor implements BeanFactoryPostProcessor{

  List<RabbitConfiguration> rabbitConfigurations;

  public IntegrationFlowPostProcessor(List<RabbitConfiguration> rabbitConfigurations) {
    this.rabbitConfigurations = rabbitConfigurations;
  }

  @Override
  public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
    rabbitConfigurations.stream()
        .forEach(rabbitConfig -> {
          IntegrationFlow intFlow = integrationFlow(amqpTemplate(rabbitConfig));
          beanFactory.registerSingleton(rabbitConfig.getHostanme(), intFlow);
        });
  }

  private AmqpTemplate amqpTemplate(RabbitConfiguration rabbitConfiguration) {
    //Implement here
    return null;
  }

  private IntegrationFlow integrationFlow(AmqpTemplate amqpTemplate) {
    //Implement here
    return null;
  }
}

然后你必须在你的配置类中注册后处理器：

@Bean public IntegrationFlowPostProcessor ifpp(List<RabbitConfiguration> config {
  return new IntegrationFlowPostProcessor(config);
}

然后，您可以使用 @Qualifier 将每个集成流按主机名注入到其他 bean 中，或者作为一个集合，比如说 List<IntegrationFlow> 以及 Spring 上下文中存在的所有集成流。

【讨论】：