Spring Integration：如何实现消息顺序处理及文件成功处理后删除

先解决测试代码的参数错误问题 #

你的source方法带有AtomicInteger参数，这不符合@InboundChannelAdapter的规则——该注解标注的方法不能有参数，Spring无法自动传入参数。修改后的正确代码如下：

@Configuration
@EnableIntegration
public class SeqChannels {

    @Bean
    public AtomicInteger integerSource() {
        return new AtomicInteger();
    }

    private final AtomicInteger integerSource;

    // 构造器注入AtomicInteger
    public SeqChannels(AtomicInteger integerSource) {
        this.integerSource = integerSource;
    }

    @InboundChannelAdapter(channel = "process", poller = @Poller(fixedDelay = "1000"))
    public Message<Integer> source() {
        return MessageBuilder.withPayload(integerSource.incrementAndGet()).build();
    }

    @ServiceActivator(inputChannel = "process", outputChannel = "delete")
    public Integer process(@Payload Integer message) {
        // 可模拟处理失败：if (message % 2 == 0) throw new RuntimeException("处理失败");
        return message;
    }

    @ServiceActivator(inputChannel = "delete")
    public void delete(@Payload Integer message) {
        System.out.println("删除消息: " + message);
    }
}

实际文件处理需求的实现方案 #

你不需要用PublishSubscribeChannel，直接用默认的DirectChannel（Spring Integration默认通道，保证消息顺序执行），将读取文件→处理文件→成功后删除文件三个步骤串联即可——只有处理步骤无异常完成，消息才会传递到删除步骤。

注解方式实现完整流 #

@Configuration
@EnableIntegration
public class FileProcessingFlow {

    // 配置文件读取适配器，监听指定目录
    @Bean
    @InboundChannelAdapter(channel = "fileInputChannel", poller = @Poller(fixedDelay = "5000"))
    public MessageSource<File> fileReadingMessageSource() {
        FileReadingMessageSource source = new FileReadingMessageSource();
        source.setDirectory(new File("你的源文件目录路径"));
        // 可选：过滤文件类型，比如只处理txt文件
        source.setFilter(new SimpleFileListFilter(f -> f.getName().endsWith(".txt")));
        return source;
    }

    // 文件处理逻辑，处理成功后传递消息到删除通道
    @ServiceActivator(inputChannel = "fileInputChannel", outputChannel = "fileDeleteChannel")
    public File processFile(@Payload File file) throws IOException {
        // 这里编写你的实际文件处理逻辑（读取、解析、存储等）
        // 若处理失败，直接抛出异常，消息将不会进入删除通道
        String content = Files.readString(file.toPath());
        System.out.println("处理文件内容: " + content);
        return file;
    }

    // 删除文件操作，仅处理成功时执行
    @ServiceActivator(inputChannel = "fileDeleteChannel")
    public void deleteFile(@Payload File file) {
        if (file.delete()) {
            System.out.println("成功删除文件: " + file.getName());
        } else {
            System.err.println("删除文件失败: " + file.getName());
        }
    }

    // 可选：全局错误处理，捕获处理失败的情况
    @Bean
    public IntegrationFlow errorHandlingFlow() {
        return IntegrationFlow.from("errorChannel")
                .handle(message -> {
                    MessagingException exception = (MessagingException) message.getPayload();
                    File failedFile = (File) exception.getFailedMessage().getPayload();
                    System.err.println("文件处理失败，不删除文件: " + failedFile.getName() + "，错误信息: " + exception.getMessage());
                })
                .get();
    }
}

核心要点说明 #

• 通道特性：默认的DirectChannel保证消息按顺序处理，且只有当前步骤执行完成（无异常），才会将消息传递到下一个组件。
• 异常控制：如果processFile抛出异常，消息会被路由到errorChannel，不会进入删除步骤，自然不会执行文件删除。
• 文件读取管控：FileReadingMessageSource默认会跟踪已处理文件，避免重复读取，可通过配置不同FileListFilter调整行为。

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