Spring Integration Java DSL中发布订阅通道的订阅者调用顺序控制

需求背景 #

从远程目录接收文件后，需要先将文件内容存入数据库，确认写入成功后再执行文件删除操作，因此必须严格控制发布订阅通道中两个订阅者的执行顺序。

参考代码 #

原示例代码如下：

@Bean
public SubscribableChannel httpInAdapterPubSubChannel() {
    return MessageChannels.publishSubscribe("httpInAdapterPubSubChannel")
            .get();
}

@Bean
public IntegrationFlow subscriber1() {
    return IntegrationFlows.from(httpInAdapterPubSubChannel())
            .handle(message -> System.out.println("根据负载丰富头信息...."))
            .get();
}

@Bean
public IntegrationFlow subscriber2() {
    return IntegrationFlows.from(httpInAdapterPubSubChannel())
            .handle(message -> System.out.println("将负载存入审计表..."))
            .get();
}

解决方案 #

默认情况下，发布订阅通道的订阅者执行顺序是不确定的，取决于Spring Bean的初始化顺序。要指定固定执行顺序，有两种可靠方式：

1. 使用@Order注解或Ordered接口

给订阅者的处理逻辑指定优先级数值，数值越小，执行顺序越靠前。

比如将“存入数据库”的订阅者设为@Order(1)，“删除文件”的订阅者设为@Order(2)：

@Bean
@Order(1) // 先执行数据库写入
public IntegrationFlow dbSaveFlow() {
    return IntegrationFlows.from(httpInAdapterPubSubChannel())
            .handle(message -> {
                // 执行数据库写入操作
                System.out.println("文件内容存入数据库完成");
            })
            .get();
}

@Bean
@Order(2) // 数据库写入成功后再执行删除
public IntegrationFlow fileDeleteFlow() {
    return IntegrationFlows.from(httpInAdapterPubSubChannel())
            .handle(message -> {
                // 执行文件删除操作
                System.out.println("文件删除完成");
            })
            .get();
}

2. 手动按顺序添加订阅者

在创建PublishSubscribeChannel时，直接调用subscribe()方法按顺序添加订阅者，这样就能严格保证执行顺序：

@Bean
public MessageHandler dbSaveHandler() {
    return message -> {
        // 数据库写入逻辑
        System.out.println("文件内容存入数据库完成");
    };
}

@Bean
public MessageHandler fileDeleteHandler() {
    return message -> {
        // 文件删除逻辑
        System.out.println("文件删除完成");
    };
}

@Bean
public SubscribableChannel fileProcessPubSubChannel() {
    PublishSubscribeChannel channel = MessageChannels.publishSubscribe("fileProcessPubSubChannel").get();
    channel.subscribe(dbSaveHandler()); // 先添加数据库订阅者
    channel.subscribe(fileDeleteHandler()); // 再添加删除文件订阅者
    return channel;
}

关键注意事项 #

• 默认的PublishSubscribeChannel是同步执行的，前一个订阅者执行完成后才会触发下一个，因此只要顺序正确，就能保证数据库写入成功后再执行删除。
• 如果给通道配置了Executor（异步模式），即使指定了顺序，也无法保证执行顺序和执行结果的依赖关系，这种场景下不建议使用异步，或者需要额外的同步机制。

内容的提问来源于stack exchange，提问作者Christoph Dahlen