Spring集成-拆分后发布订阅通道聚合

Question

我有一个基于DSL的流，它使用拆分迭代对象列表并发送Kafka消息：

.transform(...)
.split()
.channel(KAFKA_OUT_CHANNEL)

在所有消息发出后，我需要调用服务，还需要记录处理了多少消息。我知道一种方法是使用publishSubscribeChannel，其中第一个subscribe执行实际的Kafka发送，然后聚合执行服务调用：

.transform(...)
.split().
.publishSubscribeChannel(pubSub -> pubSub
        .subscribe(f -> f.channel(KAFKA_OUT_CHANNEL)))

我在弄清楚如何使用DSL在pubSubChannel中实际执行.聚合部分时遇到了问题。到目前为止，我已经尝试过：

.subscribe(f ->  f.channel(KAFKA_OUT_CHANNEL)
.subscribe(f -> f.aggregate(c -> c.processor( ?? ))))

有什么指示吗？

Answer 1

这取决于聚合后需要什么-如果您只需要一个有效负载列表，只需使用aggregate（）。。。

@SpringBootApplication
public class So51059703Application {

    public static void main(String[] args) {
        SpringApplication.run(So51059703Application.class, args);
    }

    @Bean
    public ApplicationRunner runner(ApplicationContext context) {
        return args -> {
            context.getBean("flow.input", MessageChannel.class).send(new GenericMessage<>(
                    Arrays.asList("a", "b", "c")));
        };
    }

    @Bean
    public IntegrationFlow flow() {
        return f -> f
                .split()
                .publishSubscribeChannel(p -> p
                        .subscribe(f1 -> f1.handle(System.out::println))
                        .subscribe(f2 -> f2
                                .aggregate()
                                .handle(System.out::println)));
    }

}

如果您只需要计数：

@SpringBootApplication
public class So51059703Application {

    public static void main(String[] args) {
        SpringApplication.run(So51059703Application.class, args);
    }

    @Bean
    public ApplicationRunner runner(ApplicationContext context) {
        return args -> {
            context.getBean("flow.input", MessageChannel.class).send(new GenericMessage<>(
                    Arrays.asList("a", "b", "c")));
        };
    }

    @Bean
    public IntegrationFlow flow() {
        return f -> f
                .split()
                .publishSubscribeChannel(p -> p
                        .subscribe(f1 -> f1.handle(System.out::println))
                        .subscribe(f2 -> f2
                                .aggregate(c -> c
                                        .processor(processor(), "reduce"))
                                .handle(System.out::println)));
    }

    @Bean
    public Object processor() {
        return new Object() {

            public int reduce(List<Message<?>> messages) {
                return messages.size();
            }

        };
    }

}

Answer 2

默认情况下，AbstractMessageSplitter有一个applySequence=true：

/**
 * Set the applySequence flag to the specified value. Defaults to true.
 * @param applySequence true to apply sequence information.
 */
public void setApplySequence(boolean applySequence) {

因此，我们在消息中包含以下标题：

if (this.applySequence) {
    builder.pushSequenceDetails(correlationId, sequenceNumber, sequenceSize);
}

聚合器的默认关联策略实际上基于IntegrationMessageHeaderAccessor。CORRELATION\u ID标题。这样，它将具有相同相关性键的消息收集到相同的消息组中。默认的发布策略基于消息组和sequenceSize头的比较。最后，默认的消息组处理器（MessageGroupProcessor）只需将组中的所有消息收集到一条消息中，并将收集作为有效负载。换句话说，聚合器的默认行为与拆分器完全相反。

我不知道您将从聚合器执行什么输出，但您不需要任何其他逻辑来配置聚合器-关联和释放逻辑应该基于默认状态。

您可以在参考手册中找到足够的信息。