1. 为什么文件同步这件事值得单独拎出来讲做过企业级系统集成的人都有一个共识文件同步看似简单实际上是最容易翻车的环节之一。你可能觉得不就是把A目录的文件搬到B目录吗写个定时任务加个FileUtils.copyFile不就完了我一开始也是这么想的直到在一个项目里踩了坑——对方FTP服务器网络抖动导致文件传了一半断了我们的程序认为传完了直接删了源文件结果数据丢了排查了一整天才定位到问题。从那以后但凡涉及跨系统的文件同步我再也不敢用裸写代码的方式去处理。后来接触到Apache Camel才发现这个领域早就有了非常成熟的解决方案。Camel 提供了一套基于企业集成模式EIP的路由框架把文件同步这件事拆解成了端点、路由、处理器、异常策略等标准化组件你只需要声明式地配置“从哪里取、怎么处理、到哪里去”剩下的重试、幂等、错误处理它都帮你兜住了。这篇文章要聊的就是SpringBoot 集成 Apache Camel FTP 实现文件同步的完整方案。核心关键词包括SpringBoot、Apache Camel、FTP组件、文件同步、路由配置、幂等消费、异常重试。适合谁看如果你正在做系统间文件对接、数据采集、报表分发这类需求或者你已经在用SpringBoot但还没找到优雅的文件同步方案那这篇内容应该能帮你少走不少弯路。我会从架构选型讲到代码落地再到踩坑经验尽量把每个决策背后的“为什么”说清楚。2. 整体方案设计与技术选型思路2.1 为什么是 Camel 而不是手写定时任务先说说选型这件事。文件同步的需求最直接的实现方式就是Scheduled注解加一段文件操作代码。这种方案在简单场景下确实够用但一旦需求变复杂问题就来了。我列一下手写方案和 Camel 方案的对比你感受一下对比维度手写定时任务SpringBoot Camel FTP连接管理自己维护连接池、断线重连组件内置连接管理和自动重连异常重试需要自己写重试逻辑内置 redelivery 策略配置即可幂等消费需要自己记录已处理文件支持 idempotentRepository 开箱即用文件过滤手写文件名匹配逻辑支持 include/exclude 正则表达式传输模式自己处理二进制/文本模式支持 binary/ascii 自动切换并发消费需要自己控制线程和锁支持多消费者并发配置简单监控需要自己埋点支持 JMX、事件通知你看手写方案在简单场景下确实快但每多一个需求就多一层复杂度。而 Camel 把这些都抽象成了配置项你只需要关注业务逻辑本身。2.2 Camel FTP 组件的核心概念在动手之前有几个 Camel 的核心概念必须先搞清楚不然后面看代码会懵。Endpoint端点是 Camel 与外部系统交互的入口或出口。在 FTP 场景里ftp://host:port/path就是一个端点它告诉 Camel 去哪里拉文件或者推文件。Route路由是 Camel 的核心编排单元定义了消息从起点到终点的流转路径。一个典型的路由长这样from(ftp://...).to(file://...)意思是从 FTP 拉文件写到本地目录。Processor处理器是路由中的业务逻辑节点你可以在这里对消息做转换、校验、记录日志等操作。Component组件是 Camel 对某种协议的实现封装。FTP 组件底层用的是 Apache Commons Net 库封装了 FTP 协议的连接、登录、传输、断开等操作。Consumer消费者是路由起点的角色负责从端点拉取消息。FTP 消费者会定期轮询远程目录发现有新文件就拉下来送进路由。Producer生产者是路由终点的角色负责把消息推送到目标端点。比如写到本地文件系统或者推送到另一个 FTP 服务器。2.3 整体架构设计我们这套方案的整体架构是这样的[远程FTP服务器] --(拉取)-- [Camel FTP Consumer] | v [路由处理/文件校验] | v [本地文件写入 / 转发] | v [处理结果记录/通知]核心流程分四步第一步Camel 的 FTP 消费者定时轮询远程目录发现有符合条件的文件就下载第二步文件进入路由后经过处理器做校验和转换第三步处理后的文件写到目标位置本地目录或另一个FTP第四步记录处理结果用于幂等判断和问题追溯。这个架构的关键设计点在于拉取和处理是解耦的。FTP 消费者只负责把文件拉下来具体怎么处理由路由决定。这样后面如果要改处理逻辑不需要动 FTP 配置要换 FTP 服务器也不需要改处理逻辑。3. 环境搭建与核心依赖配置3.1 项目依赖引入先看 Maven 依赖。SpringBoot 集成 Camel 需要引入 camel-spring-boot-starter 和 camel-ftp 两个核心依赖dependency groupIdorg.apache.camel.springboot/groupId artifactIdcamel-spring-boot-starter/artifactId version3.21.0/version /dependency dependency groupIdorg.apache.camel/groupId artifactIdcamel-ftp/artifactId version3.21.0/version /dependency这里有个版本选择的坑要提醒一下Camel 3.x 和 4.x 的 API 差异比较大特别是 FTP 组件的配置项命名有变化。如果你用的是 SpringBoot 2.x建议选 Camel 3.18 到 3.22 之间的版本兼容性最稳。SpringBoot 3.x 的话可以上 Camel 4.x但要注意部分配置项需要调整。另外camel-ftp这个依赖其实包含了camel-core但如果你还需要用到其他组件比如camel-jackson做JSON转换记得单独引入。3.2 FTP 连接配置连接配置我建议放在application.yml里方便不同环境切换camel: ftp: host: 192.168.1.100 port: 21 username: sync_user password: Sync2024 remote-dir: /data/upload local-dir: /data/local/sync poll-interval: 30000 delay: 5000这里每个参数都有讲究。poll-interval是轮询间隔单位毫秒设太短会给FTP服务器压力设太长文件同步延迟高。我的经验是如果对实时性要求不高30秒到1分钟比较合适如果要求准实时可以设到5到10秒但要确认FTP服务器能承受这个轮询频率。delay是首次轮询的延迟时间设个几秒是为了等应用完全启动后再去连FTP避免启动阶段的连接异常。3.3 连接池与超时参数FTP 连接有几个关键的超时参数不设的话默认值可能不适合你的网络环境camel: ftp: connect-timeout: 10000 timeout: 30000 so-timeout: 30000 maximumReconnectAttempts: 3 reconnectDelay: 5000 disconnect: trueconnect-timeout是建立连接的超时timeout是数据传输的超时so-timeout是socket读取超时。这三个参数要根据你的网络质量来调。内网环境可以设短一点比如10秒跨机房或网络不稳定的环境建议设到30秒以上。maximumReconnectAttempts和reconnectDelay控制断线重连行为。我一般设3次重试每次间隔5秒。重试次数不宜过多否则一个坏连接会拖住整个轮询周期。注意disconnect参数设为 true 表示每次消费完成后主动断开连接。这在高频轮询场景下会增加连接开销但能避免连接被FTP服务器单方面断开导致的异常。如果你的轮询间隔很短比如5秒可以设为 false 复用连接。4. 路由配置与核心代码实现4.1 基础路由从FTP拉取到本地先看最基础的路由配置把文件从FTP拉到本地目录Component public class FtpSyncRoute extends RouteBuilder { Value(${camel.ftp.host}) private String host; Value(${camel.ftp.port}) private int port; Value(${camel.ftp.username}) private String username; Value(${camel.ftp.password}) private String password; Value(${camel.ftp.remote-dir}) private String remoteDir; Value(${camel.ftp.local-dir}) private String localDir; Override public void configure() throws Exception { String ftpUrl String.format( ftp://%s:%d/%s?username%spassword%s delay%ddeletefalsebinarytrue include.*\\.csvrecursivefalse idempotenttrueidempotentKey${file:name}-${file:modified}, host, port, remoteDir, username, password, 30000); from(ftpUrl) .routeId(ftp-to-local-sync) .log(发现文件: ${file:name}, 大小: ${file:size} bytes) .to(file:// localDir ?fileName${file:name}) .log(文件同步完成: ${file:name}); } }这段代码里有几个关键配置需要展开说。deletefalse表示拉取后不删除远程文件。这个设置要慎重如果你设成 true文件拉下来后远程就删了万一本地处理失败数据就找不回来了。我一般建议先设 false等确认整个链路稳定后再考虑是否开启删除。binarytrue表示二进制传输模式。CSV、Excel、图片、压缩包这些都要用二进制模式否则文件内容会损坏。只有纯文本文件在特定编码下才考虑用 ASCII 模式。include.*\\.csv是文件过滤正则只拉取 .csv 文件。这个正则的写法要注意转义在 Java 字符串里\\.才表示正则中的\.。idempotenttrue开启幂等消费配合idempotentKey使用。${file:name}-${file:modified}这个组合键的意思是文件名加修改时间。这样同一个文件如果内容更新了修改时间变了会被重新消费如果没变就不会重复处理。4.2 幂等仓库的选型与配置上面提到的idempotenttrue默认用的是内存级别的幂等仓库应用重启后就失效了。生产环境建议换成基于文件或数据库的持久化仓库Configuration public class IdempotentConfig { Bean public IdempotentRepositoryString fileIdempotentRepository() { FileIdempotentRepository repository new FileIdempotentRepository(); repository.setFileStore(/data/camel/idempotent-repo.txt); return repository; } }然后在路由中引用这个仓库from(ftpUrl) .idempotentConsumer( header(CamelFileName), fileIdempotentRepository()) .to(file:// localDir);这里有个细节要注意FileIdempotentRepository内部用的是内存缓存加文件持久化的方式启动时会把文件里的记录加载到内存。如果记录量很大比如几十万条启动会比较慢内存占用也高。这种情况下建议用基于数据库的幂等仓库自己实现IdempotentRepository接口把已处理文件的标识存到数据库表里。4.3 文件处理器的编写文件拉下来之后通常需要做一些业务处理比如解析内容、校验格式、转换编码等。这些逻辑放在 Processor 里Component public class FileProcessProcessor implements Processor { private static final Logger log LoggerFactory.getLogger(FileProcessProcessor.class); Override public void process(Exchange exchange) throws Exception { String fileName exchange.getIn() .getHeader(Exchange.FILE_NAME, String.class); long fileSize exchange.getIn() .getHeader(Exchange.FILE_LENGTH, Long.class); InputStream body exchange.getIn() .getBody(InputStream.class); log.info(开始处理文件: {}, 大小: {} bytes, fileName, fileSize); // 校验文件大小 if (fileSize 0) { throw new IllegalArgumentException( 文件为空跳过处理: fileName); } // 校验文件扩展名 if (!fileName.endsWith(.csv)) { throw new IllegalArgumentException( 不支持的文件类型: fileName); } // 这里可以加入具体的业务处理逻辑 // 比如解析CSV、数据入库、格式转换等 exchange.getIn().setHeader(processStatus, SUCCESS); exchange.getIn().setHeader(processTime, System.currentTimeMillis()); } }Processor 里抛出的异常会被 Camel 的异常处理机制捕获触发重试或者进入死信通道。这里的关键是校验逻辑要前置不合规的文件尽早拒绝不要等到写入了目标位置才发现问题。4.4 异常处理与重试策略Camel 的异常处理是通过onException子句配置的可以针对不同类型的异常设置不同的处理策略Override public void configure() throws Exception { // 通用异常处理 onException(GenericFileOperationFailedException.class) .maximumRedeliveries(3) .redeliveryDelay(5000) .backOffMultiplier(2) .useExponentialBackOff() .handled(true) .log(FTP操作异常已重试${header.CamelRedeliveryCounter}次: ${exception.message}) .to(file:///data/camel/error?fileNameerror-${date:now:yyyyMMddHHmmss}.txt); // 业务校验异常不重试直接记录 onException(IllegalArgumentException.class) .maximumRedeliveries(0) .handled(true) .log(业务校验失败: ${exception.message}) .to(file:///data/camel/rejected); from(ftpUrl) .routeId(ftp-to-local-sync) .process(new FileProcessProcessor()) .to(file:// localDir); }backOffMultiplier(2)配合useExponentialBackOff()实现指数退避重试第一次等5秒第二次等10秒第三次等20秒。这种方式比固定间隔重试更合理因为如果是对方服务暂时不可用给它的恢复时间应该越来越长。handled(true)表示异常被当前策略处理了不会再往上抛。如果不设这个异常会继续传播可能导致路由中断。5. 进阶场景与性能优化5.1 多文件并发消费默认情况下FTP 消费者是单线程的一次只处理一个文件。如果文件量大同步速度会很慢。Camel 支持通过concurrentConsumers参数开启多消费者并发String ftpUrl String.format( ftp://%s:%d/%s?username%spassword%s delay30000concurrentConsumers5 maxMessagesPerPoll50, host, port, remoteDir, username, password);concurrentConsumers5表示开5个线程同时拉取文件。maxMessagesPerPoll50限制每次轮询最多拉50个文件避免一次拉太多把内存撑爆。但并发消费有个前提幂等仓库必须是线程安全的。FileIdempotentRepository内部用了ConcurrentHashMap基本能扛住并发场景。但如果你自己实现仓库一定要考虑线程安全问题。另外并发数不是越大越好。FTP 服务器通常对单用户的并发连接数有限制一般不超过10个。设太大反而会被服务器拒绝。我的经验是先设3到5个观察FTP服务器的负载和响应时间再决定是否往上调。5.2 大文件的分片传输如果要同步的文件很大比如几百MB的压缩包一次性传输容易超时。Camel 提供了streamDownload参数支持流式下载String ftpUrl String.format( ftp://%s:%d/%s?username%spassword%s streamDownloadtruestepwisefalse bufferSize65536, host, port, remoteDir, username, password);streamDownloadtrue表示流式下载文件不会全部加载到内存而是边下载边写入目标位置。bufferSize65536是缓冲区大小64KB 是个比较均衡的值太小会增加IO次数太大占用内存。注意流式下载模式下如果传输过程中断目标文件会是一个不完整的文件。所以一定要配合幂等机制和文件完整性校验比如比对文件大小或MD5来使用。5.3 文件完整性校验文件传输最怕的就是传了一半断了但程序以为传完了。我一般会在路由里加一道校验Component public class FileIntegrityProcessor implements Processor { Override public void process(Exchange exchange) throws Exception { Long remoteSize exchange.getIn() .getHeader(CamelFileLength, Long.class); File localFile exchange.getIn().getBody(File.class); if (remoteSize ! null localFile.length() ! remoteSize) { throw new IllegalStateException(String.format( 文件大小不一致远程: %d, 本地: %d, 文件: %s, remoteSize, localFile.length(), localFile.getName())); } } }这个校验放在写入目标文件之后执行。如果大小不一致抛出异常触发重试。重试时因为幂等仓库里没有记录上次没成功文件会被重新拉取。5.4 路由性能监控Camel 内置了 JMX 支持可以通过 JMX 查看路由的运行状态、处理消息数、失败数等指标。在application.yml里开启camel: jmx: enabled: true management: name: ftp-sync-app开启后可以用 JConsole 或 VisualVM 连接到应用查看org.apache.camel域下的 MBean。重点关注这几个指标ExchangesCompleted成功处理数、ExchangesFailed失败数、LastProcessingTime上次处理耗时。如果失败数持续增长说明有问题需要排查如果处理耗时越来越长可能是文件越来越大或者FTP服务器响应变慢。6. 常见问题与排查技巧实录6.1 FTP连接相关的问题问题一连接超时报ConnectException: Connection timed out这个最常见排查思路是先确认网络是否通telnet 一下FTP的端口再确认FTP服务器是否限制了来源IP最后检查防火墙规则。如果网络没问题但就是连不上可能是FTP服务器配置了被动模式端口范围而防火墙没放行这些端口。问题二登录失败报FTPConnectionClosedException检查用户名密码是否正确注意大小写。另外有些FTP服务器对同一个账号的并发连接数有限制如果之前的连接没正常断开新连接会被拒绝。这种情况下把disconnecttrue设上确保每次用完主动断开。问题三传输过程中断报SocketTimeoutException调大so-timeout和timeout参数。如果是大文件传输还要考虑streamDownloadtrue。另外检查FTP服务器的空闲超时设置有些服务器设的是30秒超过就断这种情况下要把客户端的超时设得比服务器短。6.2 文件重复消费的问题问题同一个文件被处理了多次先检查幂等仓库是否生效。如果用的是内存仓库应用重启后记录会丢失导致文件被重新消费。换成FileIdempotentRepository或数据库仓库可以解决。另外检查idempotentKey的配置。如果只用${file:name}作为key那么文件内容更新后不会被重新消费。如果希望内容更新后重新处理要用${file:name}-${file:modified}。还有一种情况多个应用实例同时连同一个FTP目录每个实例都有自己的幂等仓库互相不知道对方处理了哪些文件。这种场景需要用共享的幂等仓库比如基于数据库或Redis的实现。6.3 文件内容损坏的问题问题传输后的文件打不开或内容乱码首先检查binary参数是否设为 true。文本文件在ASCII模式下传输时换行符会被转换可能导致内容变化。二进制模式则原样传输。其次检查文件编码。如果源文件是GBK编码而你的程序按UTF-8读取就会乱码。这种情况下需要在处理器里做编码转换。还有一个容易忽略的点FTP服务器上的文件可能还在写入中。如果文件正在被上传你这时候去拉取拿到的就是不完整的文件。解决办法是让文件上传方先传到一个临时目录传完后再移到正式目录或者用文件名的后缀来标识比如.tmp表示未完成.csv表示已完成Camel 只拉取.csv的文件。6.4 常见问题速查表问题现象可能原因排查方向解决方案连接超时网络不通/防火墙拦截telnet测试端口检查网络和防火墙规则登录失败账号密码错误/并发限制查看FTP服务器日志核对凭据控制并发数传输中断超时设置过短检查so-timeout调大超时启用流式下载文件重复处理幂等仓库失效检查仓库类型换持久化仓库文件内容损坏传输模式错误检查binary参数设为true文件不完整源文件正在写入检查文件修改时间用临时后缀区分内存溢出大文件全量加载检查文件大小启用streamDownload路由不启动依赖缺失/配置错误查看启动日志检查依赖和配置项6.5 几个我踩过的坑坑一FTP被动模式端口问题。很多FTP服务器默认用被动模式数据连接用的端口是随机的。如果客户端在防火墙后面这些随机端口可能被拦截。解决办法是在FTP服务器上配置固定的被动端口范围然后在防火墙上放行。坑二文件名包含中文。FTP协议本身对文件名编码没有统一标准不同服务器可能用不同的编码。如果文件名有中文可能出现乱码。解决办法是在连接URL上加charsetUTF-8或者根据服务器实际编码设置。坑三路由启动顺序问题。如果路由在应用完全启动前就开始轮询FTP可能因为数据源没准备好而报错。解决办法是加delay10000让路由延迟启动或者用DependsOn控制Bean的初始化顺序。坑四日志太多。Camel 默认会打印很多DEBUG级别的日志生产环境建议把org.apache.camel的日志级别调到WARN只保留关键信息。7. 一些实战中的经验体会这套方案我在几个项目里都用过整体稳定性没问题。最后分享几个我觉得比较重要的经验。关于轮询间隔的设置我的建议是不要一上来就设得很短。先设30秒跑一段时间观察FTP服务器的负载和同步延迟如果确实需要更实时再往下调。我见过有人设成1秒的结果FTP服务器直接被轮询请求打挂了。关于幂等仓库的选择单实例部署用FileIdempotentRepository就够了多实例部署一定要用共享仓库。我一般用数据库表来实现建一个file_sync_record表字段包括文件名、修改时间、处理状态、处理时间用文件名加修改时间做唯一索引。关于异常处理不要所有异常都重试。网络类的异常连接超时、传输中断适合重试业务类的异常格式错误、校验失败重试多少次都没用直接记录到失败目录人工处理更高效。关于监控告警Camel 的 JMX 指标可以对接 Prometheus配合 Grafana 做可视化。我一般会配两个告警规则失败率超过5%告警连续5分钟没有成功处理记录也告警。后者能发现“路由卡死但没报错”这种隐蔽问题。这个方案后续还可以扩展的方向比如加上文件加密解密、对接对象存储、支持SFTP协议等。Camel 的组件生态很丰富换协议只需要改一下端点URL的前缀路由逻辑基本不用动这也是当初选它的一个重要原因。