Spring Integration的动态配置

Question

从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方法,让我AmqpTemplate从其中一个List<RabbitConfiguration>,如下:

@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();
}

然而,正是这样做的正确方法是什么,并由Spring处理List<RabbitConfiguration>结果List<IntegrationFlow>？当然不仅仅是:

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

Answer 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 上下文中存在的所有集成流。