SpringBoot 集成 Apache Camel FTP 实现文件同步
SpringBoot 集成 Apache Camel FTP 实现文件同步
1.1 导入依赖
此处最好保持 camel-spring-boot-starter 与 camel-ftp 版本一致。
<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>1.2 配置文件
1.2.1 示例
可以配置多个 ftp 同时同步。这里的 url 是为了方便显示才换的行，别傻乎乎的直接抄了。在 Camel FTP 中 URL 支持三种形式：
♞ ftp://[username@]hostname[:port]/directoryname[?options]
♞ sftp://[username@]hostname[:port]/directoryname[?options]
♞ ftps://[username@]hostname[:port]/directoryname[?options]
示例文件中的 options 配置说明：
♞ username：这个是用户名不用说了；
♞ password：这个是密码也不用说；
♞ filter：这个玩意是配置一个过滤器，用来过滤不需要下载的文件，之后详细说；
♞ recursive：是都遍历文件夹下载文件，默认不遍历，同步后会自动创建同名文件夹存放文件；
♞ idempotent：这个玩意和 noop 一起用可以启用幂等，跳过已处理文件；但是我发现没生效，可能哪里搞错了；
♞ reconnectDelay：这个玩意是延迟 n 毫秒后重连；
♞ binary：是否已二进制传输；
♞ passiveMode：是否使用被动模式；
♞ delete：同步完成后是否删除源文件；
♞ delay：间隔多少 ms 扫描一次文件夹；
♞ ftpClient.controlEncoding：配置 ftpClient 编码格式。
ftp:
img:
url: ftp://xxx.xxx.xxx.xxx:21?username=wiseftp&password=wiseftp&filter=#imgFilter
&recursive=true&reconnectDelay=1000&binary=true&passiveMode=true&delete=true&delay=500
&noop=true&idempotent=true&ftpClient.controlEncoding=GBK
# 本地下载目录
dir: file:C:\Users\softw\Desktop\ffff\img
file:
url: ftp://xxx.xxx.xxx.xxx:21?username=wiseftp&password=wiseftp&filter=#fileFilter
&recursive=true&reconnectDelay=1000&binary=true&passiveMode=true&delete=false&delay=5000
&noop=true&idempotent=true&ftpClient.controlEncoding=GBK
# 本地下载目录
dir: file:C:\Users\softw\Desktop\ffff\file
# 后台运行进程
camel:
springboot:
main-run-controller: true1.2.2 配置列表
以下配置只列举了常用或者我能够理解的配置，其他配置请查看 ☞ 官方文档
名称 | 说明 | 默认值 |
|---|---|---|
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 毫秒，然后再执行重新连接尝试 |
1.3 配置路由
1.3.1 xml 配置
使用这种方式配置路由需要再启动类中使用 @ImportResource(locations = {"classpath:camel.xml"}) 加载配置文件，可以配置多个路由。若将 from 配置为本地地址，to 配置为远端地址，则可以实现向远端服务器上传文件。
<route>
<from uri="ftp://xxxxxx"/>
<to uri="file://xxxx/xxxx"/>
</route>1.3.2 配置类
使用配置类则需要继承 RouteBuilder 类，然后重写 configure() 方法
/**
* Created with IntelliJ IDEA.
*
* @author Demo_Null
* @date 2020/8/13
* @description ftp 路由配置
*/
@Component
public class FtpRouteBuilder extends RouteBuilder {
@Value("${ftp.img.url}")
private String sftpServerImg;
@Value("${ftp.img.dir}")
private String downloadLocationImg;
@Value("${ftp.file.url}")
private String sftpServerFile;
@Value("${ftp.file.dir}")
private String downloadLocationFile;
@Autowired
private DataProcessor dataProcessor;
@Override
public void configure() throws Exception {
// from 内为 URL
from(sftpServerImg)
// 本地路径
.to(downloadLocationImg)
// 数据处理器
.process(dataProcessor)
// 日志
.log(LoggingLevel.INFO, logger, "Download img ${file:name} complete.");
from(sftpServerFile)
.to(downloadLocationFile)
.log(LoggingLevel.INFO, logger, "Download file ${file:name} complete.");
}
}1.4 过滤器
当 FTP 服务器上有许多文件，但是我们只需要 .jpg 文件的时候可以使用 camel-ftp 的文件过滤器来实现，在 url 中的 filter 来指定使用那个过滤器，例如 filter=#imgFilter；自定义的过滤器需要实现GenericFileFilter 接口并重写 accept 方法。若是需要遍历文件夹，则此处应该对于文件夹直接放行。
/**
* Created with IntelliJ IDEA.
*
* @author Demo_Null
* @date 2020/8/13
* @description 过滤器
*/
@Component
public class ImgFilter implements GenericFileFilter<Object> {
@Value("${ftp.img.dir}")
private String fileDir;
@Override
public boolean accept(GenericFile<Object> genericFile) {
return genericFile.getFileName().endsWith(".jpg") || genericFile.isDirectory();
}
}1.5 文件处理器
在进行文件下载时我们可能需要改变下载文件的存储目录，或者进行入库等操作。这时就需要文件处理器对于下载的文件进行处理，一个文件处理器需要实现 Processor 接口并重写 process(Exchange exchange) 方法。
/**
* Created with IntelliJ IDEA.
*
* @author Demo_Null
* @date 2020/8/13
* @description 文件处理器
*/
@Component
public class DataProcessor implements Processor {
@Value("${ftp.img.dir}")
private String fileDir;
@Autowired
private JdbcTemplate jdbcTemplate;
@Override
@Transactional
public void process(Exchange exchange) throws Exception {
try {
GenericFileMessage<RandomAccessFile> inMsg = (GenericFileMessage<RandomAccessFile>) exchange.getIn();
String fileName = inMsg.getGenericFile().getFileName();
String sql = "insert into file(filename) values (?)";
jdbcTemplate.update(sql, fileName);
} catch (Exception e) {
e.printStackTrace();
}
}
}