java - Spring 集成 channel 链怪异

Question

我可能只是在这里遗漏了一些非常简单的东西(或误用了一些东西)，但我试图设置两个直接 channel ，以便一个流将一些数据按顺序传递给每个 channel 。因此，使用 Spring Integration JAVA DSL，我得到了类似的东西(本示例已大大简化):

   public static final String TEST_CHANNEL = "testGateway";
public static final String TEST_UPPER_CHANNEL = "testChannelUpper";
public static final String TEST_LOWER_CHANNEL = "testChannelLower";

@Bean(name = TEST_CHANNEL)
public MessageChannel testGatewayChannel() {
    return MessageChannels.direct(TEST_CHANNEL).get();
}

@Bean(name = TEST_UPPER_CHANNEL)
public MessageChannel testChannelUpperChannel() {
    return MessageChannels.direct(TEST_UPPER_CHANNEL).get();
}

@Bean(name = TEST_LOWER_CHANNEL)
public MessageChannel testChannelLowerChannel() {
    return MessageChannels.direct(TEST_LOWER_CHANNEL).get();
}


@Bean
public IntegrationFlow testFlow() {
    return IntegrationFlows
            .from(TEST_CHANNEL)
            .channel(TEST_UPPER_CHANNEL)
            .channel(TEST_LOWER_CHANNEL)
            .get();
}


@Bean
public IntegrationFlow testUpperFlow() {
    return IntegrationFlows
            .from(TEST_UPPER_CHANNEL)
            .<String, String>transform(String::toUpperCase)
            .handle(System.out::println)
            .get();
}

@Bean
public IntegrationFlow testLowerFlow() {
    return IntegrationFlows
            .from(TEST_LOWER_CHANNEL)
            .<String, String>transform(String::toLowerCase)
            .handle(System.out::println)
            .get();
}

我使用 REST 端点通过网关调用流，但当我这样做时，似乎只调用了其中一个 channel 。该 channel 在调用之间似乎也是随机的(有时转到 testChannelUpper，有时转到 testChannelLower)。

我基本上在执行过程中得到了这样的结果:(每次我刚刚到达这个端点 http://localhost:9090/test?test=HellOoi )

执行 1:GenericMessage [payload=HELLOOI， header ={jobName=someActivity，历史记录=someGateway，testGateway，testChannelUpper，testUpperFlow.channel#0，id=4aa7b075-23cc-6ab3-10a1-c7cb73bae49b，时间戳=1447686848477}]

执行2:GenericMessage [payload=HELLOOI， headers={jobName=someActivity，history=someGateway，testGateway，testChannelUpper，testUpperFlow.channel#0，id=a18dcd01-da18-b00d-30c0-e1a03ce19104，timestamp=1447686853549}]

执行3:GenericMessage [payload=hellooi， headers={jobName=someActivity，history=someGateway，testGateway，testChannelUpper，testLowerFlow.channel#0，id=5f0abcb9-378e-7a3c-9c93-a04ff6352927，timestamp=1447686857545}]

我相信我在这里尝试的内容也显示在 DSL wiki 的 channelFlow 示例中: https://github.com/spring-projects/spring-integration-java-dsl/wiki/Spring-Integration-Java-DSL-Reference

我正在使用的规范是:

Spring Boot v1.2.2.RELEASE

Spring v4.1.5.RELEASE

spring-integration-java-dsl 1.0.2.RELEASE

JDK 1.8.0_40-b25

那么...还有其他人见过这种行为吗？我只是滥用 channel 实现吗？还有其他想法吗？提前致谢!

<小时/>

正如 Gary 指出的，最好的方法是建立一个 pub-sub 并以此向消费者订购:

    @Bean(name = TEST_CHANNEL)
public MessageChannel testGatewayChannel() {
    return MessageChannels.publishSubscribe(TEST_CHANNEL).get();
}


@Bean
public IntegrationFlow testUpperFlow() {
    return IntegrationFlows
            .from(TEST_CHANNEL)
            .<String, String>transform(String::toUpperCase, e -> e.order(1))
            .handle(System.out::println)
            .get();
}

@Bean
public IntegrationFlow testLowerFlow() {
    return IntegrationFlows
            .from(TEST_CHANNEL)
            .<String, String>transform(String::toLowerCase, e -> e.order(2))
            .handle(System.out::println)
            .get();
}

Answer 1

这样做的目的是什么...

@Bean
public IntegrationFlow testFlow() {
    return IntegrationFlows
        .from(TEST_CHANNEL).fixedSubscriberChannel()
        .channel(TEST_UPPER_CHANNEL)
        .channel(TEST_LOWER_CHANNEL)
        .get();
}

？

所做的只是将三个 channel 桥接在一起。

事实上，您最终在 TEST_UPPER_CHANNEL 上有 2 个使用者 - 此流中的桥接器和其他流中的变压器。

默认情况下，直接 channel 中的调度使用循环分配。因此，第一条消息将发送至桥接器，第二条消息将发送至变压器，等等。