如何在Camel FTP组件中实现可靠错误处理，避免JMS重复投递

核心矛盾在于：FTP组件的文件拉取-处理-重命名流程与JMS发送不在统一的事务控制范围内。当JMS发送成功但FTP重命名到completed目录失败时，Camel默认会重试整个路由流程，导致文件被重复拉取投递，进而引发下游重复数据问题。同时当前的springTransactionErrorHandler和doTry/doCatch未覆盖FTP组件的后置处理阶段异常（即路由逻辑执行完后的文件重命名操作），导致错误消息无法进入DLQ。

1. 重构路由错误处理逻辑 #

使用onException替代doTry/doCatch，因为前者能覆盖FTP组件在路由执行完成后的后置操作异常（如重命名失败）。同时配置FTP的错误目录、幂等机制，确保文件仅被处理一次。

from("ftp://user@ftp-server/source?password=xxx" +
        "&move=completed/${date:now:yyyyMMdd}" +  // 成功后移动到completed目录
        "&moveFailed=failed/${date:now:yyyyMMdd}" +  // 失败后移动到failed目录
        "&idempotent=true" +  // 开启幂等，避免重复拉取
        "&idempotentRepository=#fileIdempotentRepo")
    .routeId("ftp-to-jms-route")
    // 生成文件唯一标识，用于幂等校验和DLQ标记
    .setHeader("FileUniqueId", simple("${file:name}-${file:size}-${file:lastModified}"))
    // JMS发送开启事务，确保消息要么投递成功，要么回滚
    .to("jms:queue:target-queue?transacted=true")

    // 捕获所有关键异常：FTP操作失败、JMS发送失败
    .onException(GenericFileOperationFailedException.class, JMSException.class)
        .handled(true)  // 标记异常已处理，终止路由重试
        .log(LoggingLevel.ERROR, "文件${file:name}处理失败，已发送至DLQ，错误信息：${exception.message}")
        .to("jms:queue:error-dlq")  // 发送错误消息到DLQ
        .process(exchange -> {
            // 手动标记文件为已处理，强制FTP组件不再重试拉取
            GenericFile<?> file = exchange.getIn().getBody(GenericFile.class);
            if (file != null) {
                exchange.setProperty(Exchange.FILE_MARKER, GenericFileOperation.COMPLETE);
            }
        })
    .end();

2. 配置持久化幂等仓库 #

生产环境必须使用持久化幂等仓库，避免服务重启后丢失已处理文件的记录，防止重复拉取。

@Bean
public IdempotentRepository fileIdempotentRepo() {
    // 生产环境建议用JDBC或Redis实现，以下为内存版示例
    return MemoryIdempotentRepository.memoryIdempotentRepository(10000);
}

3. 调整事务错误处理器 #

由于FTP不支持分布式事务，采用本地事务+补偿机制：

• JMS发送使用本地事务，确保消息投递的原子性；
• FTP重命名失败时，通过moveFailed自动将文件转移到错误目录，结合幂等仓库阻止重复拉取。

模拟FTP重命名失败场景，验证文件流向、JMS投递次数和DLQ消息：

def "测试FTP重命名失败时消息仅投递一次并进入DLQ"() {
    given:
    // 初始化MockFtpServer，模拟权限问题导致completed目录不可写
    def mockFtpServer = new MockFtpServer()
    mockFtpServer.addUser(new UserAccount("user", "password", "/ftp-root"))
    
    // 创建源目录并上传测试文件
    def sourceDir = new FileSystem("/ftp-root/source")
    sourceDir.createDirectory()
    sourceDir.createFile("test-data.txt").setContent("sample content")
    
    // 创建completed目录但设置为不可写，触发重命名失败
    def completedDir = new FileSystem("/ftp-root/completed")
    completedDir.createDirectory()
    completedDir.setWritable(false)
    
    mockFtpServer.start()

    when:
    camelContext.startRoute("ftp-to-jms-route")
    Thread.sleep(5000)  // 等待路由处理完成

    then:
    // 验证目标队列仅收到一次消息
    def targetMsg = jmsTemplate.receive("target-queue", 1000)
    targetMsg != null
    jmsTemplate.receive("target-queue", 1000) == null

    // 验证DLQ收到错误消息
    def dlqMsg = jmsTemplate.receive("error-dlq", 1000)
    dlqMsg != null
    dlqMsg.getStringProperty("FileUniqueId") == "test-data.txt-13-1699999999000"

    // 验证文件被转移到failed目录
    def failedDir = new FileSystem("/ftp-root/failed")
    failedDir.exists()
    failedDir.listFiles().size() == 1
    failedDir.listFiles()[0].getName() == "test-data.txt"

    cleanup:
    mockFtpServer.stop()
    camelContext.stopRoute("ftp-to-jms-route")
}

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