Spring Integration实现XML文件校验归档/错误目录迁移异常排查

我尝试实现一个Spring Integration类，读取并解析.xml文件，若文件校验有效则将其移动至“archived”目录，无效则移动至错误目录。实现代码如下：

import com.nagarro.studentapi.integration.queue.StudentSender;
import com.nagarro.studentapi.util.XmlParser;
import org.aopalliance.aop.Advice;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.FileWritingMessageHandler;
import org.springframework.integration.file.filters.SimplePatternFileListFilter;
import org.springframework.integration.file.support.FileExistsMode;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

import java.io.File;

@Configuration
@EnableIntegration
public class IntegrationConfiguration {

    private static final String XML = "*.xml";
    private static final String STUDENT = "\\student.xml";

    @Value("${student-api.xmlPath}")
    private String inputPath;
    @Value("${student-api.archivedDestination}")
    private String successPath;
    @Value("${student-api.errorDestination}")
    private String errorPath;

    @Bean
    public MessageChannel messageChannel() {
        return new DirectChannel();
    }

    @Bean
    @InboundChannelAdapter(value = "messageChannel")
    public MessageSource<File> messageProducer() {
        FileReadingMessageSource messageSource = new FileReadingMessageSource();
        messageSource.setDirectory(new File(inputPath));
        messageSource.setFilter(new SimplePatternFileListFilter(XML));
        return messageSource;
    }

    @Bean
    @ServiceActivator(inputChannel = "messageChannel")
    public MessageHandler handler() {
        FileWritingMessageHandler handler = new FileWritingMessageHandler(new File(successPath));
        handler.setFileExistsMode(FileExistsMode.REPLACE);
        handler.setExpectReply(false);
        return handler;
    }

    @Bean
    public IntegrationFlow integrationFlow(XmlParser xmlParser) {
        return IntegrationFlows.from(messageProducer(), spec -> spec.poller(Pollers.fixedDelay(1000)))
                .enrichHeaders(h -> h.headerExpression(FileHeaders.ORIGINAL_FILE, "payload"))
                .convert(String.class)
                .transform((String path) -> xmlParser.parsePath(path))
                .handle("xmlParser", "parsePath", e -> e.advice(errorAdvice()))
                .get();
    }

    @Bean
    public AbstractRequestHandlerAdvice errorAdvice() {
        return new AbstractRequestHandlerAdvice() {

            @Override
            protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
                File file = message.getHeaders().get(FileHeaders.ORIGINAL_FILE, File.class);
                try {
                    Object result = callback.execute();
                    file.renameTo(new File(successPath, STUDENT));
                    System.out.println("File renamed after success");
                    return result;
                }
                catch (Exception e) {
                    file.renameTo(new File(errorPath, STUDENT));
                    System.out.println("File renamed after failure");
                    throw e;
                }
            }
        };
    }
}

但调用callback.execute()时出现如下错误：

2022-09-06 18:20:07.971 ERROR 32152 --- [   scheduling-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageHandlingException: error occurred during processing message in 'MethodInvokingMessageProcessor' [org.springframework.integration.handler.MethodInvokingMessageProcessor@1135e3d6]; nested exception is java.lang.IllegalArgumentException: No candidate methods found for messages., failedMessage=GenericMessage [payload=Student(firstname=John, lastname=Dose, cnp=123, birthDate=2000-12-12, address=Address(street=a, number=1, city=Craiova, country=Romania), grades=[Grade(discipline=a, date=2021-12-12, grade=10), Grade(discipline=b, date=2021-12-12, grade=9)]), headers={....

尽管已定义消息处理器，我怀疑问题源于未重写handle方法，但不确定具体如何操作，请求帮助排查并解决该问题。

问题分析与修复 #

核心问题定位 #

错误提示No candidate methods found for messages的根本原因是：
你在IntegrationFlow中连续调用了两次xmlParser.parsePath：

1. 先通过.transform((String path) -> xmlParser.parsePath(path))将文件路径转换为Student对象
2. 接着又调用.handle("xmlParser", "parsePath")，此时消息payload已经是Student对象，而XmlParser的parsePath方法仅接受字符串类型的路径参数，找不到匹配的重载方法，因此抛出异常。

同时代码中存在冗余定义：@InboundChannelAdapter和@ServiceActivator声明的通道与处理器，和IntegrationFlow DSL定义的流程重复，会导致消息分流冲突。

修复后的完整代码 #

import com.nagarro.studentapi.util.XmlParser;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.FileWritingMessageHandler;
import org.springframework.integration.file.filters.SimplePatternFileListFilter;
import org.springframework.integration.file.support.FileExistsMode;
import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;

import java.io.File;

@Configuration
@EnableIntegration
public class IntegrationConfiguration {

    private static final String XML = "*.xml";

    @Value("${student-api.xmlPath}")
    private String inputPath;
    @Value("${student-api.archivedDestination}")
    private String successPath;
    @Value("${student-api.errorDestination}")
    private String errorPath;

    @Bean
    public FileReadingMessageSource messageProducer() {
        FileReadingMessageSource messageSource = new FileReadingMessageSource();
        messageSource.setDirectory(new File(inputPath));
        messageSource.setFilter(new SimplePatternFileListFilter(XML));
        return messageSource;
    }

    @Bean
    public IntegrationFlow integrationFlow(XmlParser xmlParser) {
        return IntegrationFlows.from(messageProducer(), spec -> spec.poller(Pollers.fixedDelay(1000)))
                // 保留原始文件信息到Header
                .enrichHeaders(h -> h.header(FileHeaders.ORIGINAL_FILE, m -> m.getPayload()))
                // 将File对象转换为绝对路径字符串
                .transform(File::getAbsolutePath)
                // 解析XML为Student对象，解析失败会抛出异常
                .transform(xmlParser::parsePath)
                // 处理成功时移动原始文件到归档目录
                .handle(m -> {
                    File originalFile = m.getHeaders().get(FileHeaders.ORIGINAL_FILE, File.class);
                    FileWritingMessageHandler successHandler = new FileWritingMessageHandler(new File(successPath));
                    successHandler.setFileExistsMode(FileExistsMode.REPLACE);
                    successHandler.setDeleteSourceFiles(true); // 写入后删除源文件，实现移动效果
                    successHandler.handleMessage(m);
                    return null;
                }, e -> e.advice(errorHandlingAdvice()))
                .get();
    }

    @Bean
    public ExpressionEvaluatingRequestHandlerAdvice errorHandlingAdvice() {
        ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
        // 异常触发时，调用错误目录的文件处理器
        advice.setOnFailureExpressionString("@errorFileHandler.handleMessage(#root)");
        advice.setTrapException(false); // 保留异常以便日志记录
        return advice;
    }

    @Bean
    public FileWritingMessageHandler errorFileHandler() {
        FileWritingMessageHandler handler = new FileWritingMessageHandler(new File(errorPath));
        handler.setFileExistsMode(FileExistsMode.REPLACE);
        handler.setDeleteSourceFiles(true);
        handler.setExpectReply(false);
        return handler;
    }
}

关键优化点 #

1. 移除冗余代码：删除了重复的MessageChannel、@InboundChannelAdapter和@ServiceActivator定义，统一使用DSL管理完整流程。
2. 修正重复解析问题：仅在transform阶段调用一次xmlParser.parsePath，避免payload类型不匹配导致的方法查找失败。
3. 标准化文件移动：使用Spring Integration原生的FileWritingMessageHandler配合setDeleteSourceFiles(true)实现文件移动，避免手动renameTo的跨系统兼容性问题。
4. 优雅异常处理：通过ExpressionEvaluatingRequestHandlerAdvice统一管理成功/失败分支，代码结构更清晰。

内容的提问来源于stack exchange，提问作者Ștefan Jiroveanu