如何在拉取完所有SFTP文件后自动停止InboundChannelAdapter轮询

问题根因 #

你自定义的停止逻辑没有生效，核心有三点问题：

1. 代码存在空指针异常：afterReceive方法中，先判断message == null后又调用了message.getPayload()，直接触发NPE，导致停止命令永远无法发出
2. 自定义通知未加入轮询切面链：你配置的PollerMetadata的adviceChain只包含了RotatingServerAdvice，没有加入自己实现的StopPollingAdvice，逻辑根本不会执行
3. 停止判断逻辑不合理：单次轮询返回空就停止，会出现第一个目录拉完就停，漏掉第二个目录文件的问题。因为你配置RotatingServerAdvice每次轮询切换目录，需要连续两个目录轮询都返回空，才代表所有存量文件已经拉取完成

解决方案 #

1. 修复停止通知的空指针与判断逻辑 #

@Component
public class StopPollingAdvice implements ReceiveMessageAdvice {
    @Autowired
    private MessageChannel controlChannel;
    // 和你配置的KeyDirectory数量保持一致，当前是2个目录
    private static final int ALL_DIR_EMPTY_COUNT = 2;
    private final AtomicInteger emptyPollCounter = new AtomicInteger(0);

    @Override
    public Message<?> afterReceive(Message<?> message, Object source) {
        if (message == null) {
            int currentEmpty = emptyPollCounter.incrementAndGet();
            // 所有目录连续轮询都为空，说明存量文件已拉完
            if (currentEmpty >= ALL_DIR_EMPTY_COUNT) {
                // 重置计数器，方便下次启动时状态正常
                emptyPollCounter.set(0);
                controlChannel.send(MessageBuilder.withPayload("@inboundtest.stop()").build());
            }
        } else {
            // 只要拿到文件就重置计数器，避免中间出现空轮询误判
            emptyPollCounter.set(0);
        }
        return message;
    }
}

2. 将停止通知加入轮询切面链 #

修改PollerMetadata的配置，按顺序加入切面：

@Bean
public PollerMetadata fileReadingMessageSourcePollerMetadata(StopPollingAdvice stopPollingAdvice) {
    PollerMetadata meta = new PollerMetadata();
    meta.setTrigger(new PeriodicTrigger(1000));
    // 先执行目录切换逻辑，再执行停止判断
    meta.setAdviceChain(List.of(advice(), stopPollingAdvice));
    meta.setMaxMessagesPerPoll(1);
    meta.setErrorHandler(throwable -> new IOException());
    return meta;
}

可选优化 #

如果后续你会动态调整轮询目录的数量，可以把ALL_DIR_EMPTY_COUNT改成配置项，和RotatingServerAdvice中keyDirectories的列表长度保持一致即可。

内容的提问来源于stack exchange，提问作者Abraham Macias