Spring集成：动态注册IntegrationFlow的刷新/重新加载

Question

我从“dynconfigflows”-目录中加载xml配置文件，然后动态创建IntegrationFlow。
每个xml文件一个IntegrationFlow。如果配置文件已编辑，则应更新IntegrationFlow。IntegrationFlow的注册id被设置为配置文件名。
，因此当前的流应该平稳地停止（没有更多的输入消息和当前处理的消息应该完成）。在此之后，应该删除它，并注册流的更新版本。
现在的问题是：
a)在下面的JUnit测试中，如果我没有调用Registration.remove(ID)，我可以用相同的ID注册两个流；这被认为是行为吗？ID不应该是唯一的吗？
b)在调用remove(...)之后，我试图向removed InputChannel再发送一条消息。我得到了一个错误，但仍然可以看到工作的removed流的日志输出。

我应该如何实现动态IntegrationFlow的刷新/重新加载功能？

@Test
public void testRegisterUnregister() {
    final String FLOW_ID = "regFlow";
    Flux<Message<?>> messageFlux = Flux.just("1,2,3,4").map(v -> v.split(",")).flatMapIterable(Arrays::asList)
            .map(Integer::parseInt).map(GenericMessage<Integer>::new);

    QueueChannel resultChannel = new QueueChannel();

    IntegrationFlow regFlow = createFlow("first", messageFlux, resultChannel);

    IntegrationFlowRegistration register1 = this.flowContext.registration(regFlow).id(FLOW_ID).register();
    final String FIRST_ID = register1.getId();
    System.out.println("!!!!!!!!!!!!!!!! register flow: "+FIRST_ID);
    // validate incoming of 4 integers
    assertThat(resultChannel.getQueueSize()).as("queueSize").isEqualTo(4);
    // save reference to first InputChannel
    MessageChannel input1 = register1.getInputChannel();

    // stop and remove first flow
    register1.stop();
    this.flowContext.remove(FIRST_ID);
    // wait for the case if flows shutdown needs some time
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    // create second flow
    regFlow = createFlow("second", messageFlux, resultChannel);
    IntegrationFlowRegistration register2 = this.flowContext.registration(regFlow).id(FLOW_ID).register();
    System.out.println("!!!!!!!!!!!!!!!! register flow: "+register2.getId());
    // send extra message to the second flow
    register2.getInputChannel().send(MessageBuilder.withPayload("!!!!!!!!!!! new registered flow message").build());
    try {
        // send extra message to the first flow. It should fail because first flow is removed.
        // It fails but we still see "!!!!!!!!!!! new registered flow message" from first flow. It should't happen!
        input1.send(MessageBuilder.withPayload("!!!!!!!!!!! new registered flow message").build());
        Assert.fail("we should get an exception because first flow is removed");
    } catch (MessageDeliveryException e) {
        e.printStackTrace();
    }

    assertThat(resultChannel.getQueueSize()).as("queueSize=2 x 4 + 1").isEqualTo(9);
}

private IntegrationFlow createFlow(String markerText, Flux<Message<?>> messageFlux, QueueChannel resultChannel) {
    return IntegrationFlows
            .from(messageFlux)
            .log(l -> "!!!!!!!!!!!!!!!!!!!!!!! "+markerText+"="+l)
            .channel(resultChannel)
            .get();
}

Answer 1

第一个问题肯定是一个bug:https://jira.spring.io/browse/int-4413。

我们可以在应用程序上下文中重写现有bean，因为默认情况下允许：

if (oldBeanDefinition != null) {
        if (!isAllowBeanDefinitionOverriding()) {
            throw new BeanDefinitionStoreException(beanDefinition.getResourceDescription(), beanName,
                    "Cannot register bean definition [" + beanDefinition + "] for bean '" + beanName +
                    "': There is already [" + oldBeanDefinition + "] bound.");
        }

但是，在integrationflowcontext中。我们将在提到的JIRA中拒绝这种状态。