1. 依赖
<dependency> <groupId>org.apache.camel.springboot</groupId> <artifactId>camel-spring-boot-starter</artifactId> <version>3.4.2</version> </dependency> <dependency> <groupId>org.apache.camel</groupId> <artifactId>camel-ftp</artifactId> <version>3.4.2</version> </dependency> 版本保持一致，可用最新版。
2. 配置文件
camel: springboot: main-run-controller: true management: endpoint: camelroutes: enabled: true read-only: true # delay=5s 间隔多少 ms 轮询一次 # readLock 读锁 # noop 如果 noop = true，Camel 也将设置 idempotent = true 以避免一遍又一遍地使用相同的文件 # recursive 是否遍历文件夹处理文件 ftp: server: info: ftp://ftpuser@172.16.113.128:21?password=123456&delay=5s&readLock=rename&ftpClient.controlEncoding=UTF-8&noop=true&recursive=true local: dir: file:D:/temp/ftp/
2.1 配置文档#
| 名称 | 说明 | 默认值 |
| username | 登录用户名 | |
| password | 登录密码 | |
| binary | 指定文件传输模式，BINARY 或 ASCII | false(ASCII) |
| charset | 指定文件读取的编码格式，写出时也可配置 | |
| disconnect | 使用后是否立即断开连接 | false |
| passiveMode | 是否启用被动模式 | false |
| separator | 设置路径分隔符，可选：UNIX，Windows，Auto | UNIX |
| delete | 是否在文件处理完成后删除源文件 | false |
| noop | 如果 noop = true，Camel 也将设置 idempotent = true 以避免一遍又一遍地使用相同的文件 | false |
| recursive | 是否遍历文件夹处理文件 | false |
| download | 是否下载文件，如果将此选项设置为 false，则消息正文将为 null | |
| filter | 设置过滤器 | |
| idempotent | 选择使用幂等消费者 EIP 模式让 Camel 跳过已处理的文件。如果 noop = true，则将启用幂等，以避免重复使用相同的文件。 | |
| delay | 间隔多少 ms 轮询一次 | |
| reconnectDelay | 延迟 n 毫秒，然后再执行重新连接尝试 |
3. 编写一个Component
继承RouteBuilder,实现configure方法。
设置数据源、目标，并且可以根据需要额外设置 过滤器Predicate和处理器Processor。
package com.example.goodsync.ftpdemo; import org.apache.camel.LoggingLevel; import org.apache.camel.builder.RouteBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; /** * @className: DownloadRouteDemo * @description: DownloadRouteDemo */ @Component public class DownloadRouteDemo extends RouteBuilder { private static Logger logger = LoggerFactory.getLogger(DownloadRouteDemo.class); @Value("${ftp.server.info}") private String ftpServer; @Value("${ftp.local.dir}") private String downloadLocation; @Autowired LocationFileProcessor locationFileProcessor; @Autowired FtpDownloadFileFilter ftpDownloadFileFilter; @Override public void configure() throws Exception { from(ftpServer) //数据源 .filter(ftpDownloadFileFilter) //过滤器 .to(downloadLocation) //目标 .log(LoggingLevel.INFO, logger, "Downloaded file ${file:name} complete.") .process(locationFileProcessor); //处理器 } }
4. 过滤器设置
过滤器设置，只处理.txt或者.json文件。上述代码中 filter 必须再to 前面才能生效。
package com.example.goodsync.ftpdemo; import org.apache.camel.Exchange; import org.apache.camel.Predicate; import org.apache.camel.component.file.GenericFileMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; import java.io.RandomAccessFile; /** * @className: FtpDownloadFileFilter * @description: FtpDownloadFileFilter */ @Component public class FtpDownloadFileFilter implements Predicate { private static Logger logger = LoggerFactory.getLogger( FtpDownloadFileFilter.class ); @Override public boolean matches(Exchange exchange) { boolean bMatches = true; GenericFileMessage<RandomAccessFile> inFileMessage = (GenericFileMessage<RandomAccessFile>) exchange.getIn(); String fileName = inFileMessage.getGenericFile().getFileName(); bMatches = fileName.endsWith(".txt") || fileName.endsWith(".json"); logger.info("fileName:"+fileName+"matches:"+bMatches); return bMatches; } }
5. 处理器
代码如下:
package com.example.goodsync.ftpdemo; import org.apache.camel.Exchange; import org.apache.camel.Processor; import org.apache.camel.component.file.GenericFileMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import java.io.RandomAccessFile; /** * @className: LocationFileProcessor * @description: LocationFileProcessor */ @Component public class LocationFileProcessor implements Processor { private static Logger logger = LoggerFactory.getLogger( LocationFileProcessor.class ); @Value("${ftp.local.dir}") private String fileDir; @Autowired OrderService orderService; @Override public void process(Exchange exchange) throws Exception { GenericFileMessage<RandomAccessFile> inFileMessage = (GenericFileMessage<RandomAccessFile>) exchange.getIn(); //文件名 String fileName = inFileMessage.getGenericFile().getFileName(); //文件的绝对路径 logger.info(fileDir + fileName); //解析入库等操作 orderService.process(fileDir + fileName); } }
6. 定义一个OrderService
package com.example.goodsync.ftpdemo; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Service; /** * @className: OrderService * @description: OrderService */ @Service public class OrderService { private static Logger logger = LoggerFactory.getLogger( OrderService.class ); public void process(String path){ logger.info("path:"+path); } }
7. 问题
7.1 更多的文档信息移步官方文档#
7.2 出现“550 create directory operation failed”错误#
原因是SELinux(Security-Enhanced Linux----是美国国家安全局对于强制访问控制的实现, 是Linux 上最杰出的新安全子系统.)安装机制搞的鬼.只要disable SELinux就可以了.
解决办法: 关闭SELinux
# vim /etc/selinux/config
将 SELINUX=XXX -->XXX 代表级别 改为 SELINUX=disabled
重启系统
7.3 文件名出现乱码#
地址后面拼接 ftpClient.controlEncoding：配置 ftpClient 编码格式。 比如ftpClient.controlEncoding=UTF-8
