Apache Camel 与 Spring Boot 集成，通过FTP定时采集、处理文件
一、maven项目需要引入相关包
pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.5.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.lvyuanj.core</groupId>
<artifactId>camel</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>camel</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<!-- camel 相关包 -->
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-spring-boot-starter</artifactId>
<version>3.0.0-RC3</version>
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-ftp-starter</artifactId>
<version>3.0.0-RC3</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.16.12</version>
</dependency>
<!-- camel 内部使用切面代理，不引入会报proxy异常 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
二、编写一个Component，扩展RouteBuilder.class,重新实现configure方法。设置数据源、目标，并且可以根据需要额外设置 过滤器Predicate和处理器Processor。
package com.lvyuanj.core.camel.service;
import lombok.extern.slf4j.Slf4j;
import org.apache.camel.LoggingLevel;
import org.apache.camel.builder.RouteBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class DownloadRouteDemo extends RouteBuilder {
@Value("${ftp.server.info}")
private String sftpServer;
@Value("${ftp.local.dir}")
private String downloadLocation;
@Override
public void configure() throws Exception {
from( sftpServer ).to( downloadLocation ).log(LoggingLevel.INFO, log, "Downloaded file ${file:name} complete.");
}
}
package com.lvyuanj.core.camel.service;
import lombok.extern.slf4j.Slf4j;
import org.apache.camel.LoggingLevel;
import org.apache.camel.builder.RouteBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class LocalTransformRoute extends RouteBuilder {
@Value("${route.parserfile.info}")
private String location;
@Value("${route.parserfile.dir}")
private String locationDir;
@Autowired
LocationFileProcessor locationFileProcessor;
@Override
public void configure() throws Exception {
from( location ).process( locationFileProcessor ).to( locationDir ).log(LoggingLevel.INFO, log, "tirans file ${file:name} complete.");
}
}
处理器代码如下:
package com.lvyuanj.core.camel.service;
import lombok.extern.slf4j.Slf4j;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.component.file.GenericFileMessage;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.io.File;
import java.io.RandomAccessFile;
@Slf4j
@Component
public class LocationFileProcessor implements Processor {
@Value("${ftp.local.dir}")
private String fileDir;
@Override
public void process(Exchange exchange) throws Exception {
GenericFileMessage<RandomAccessFile> inFileMessage = (GenericFileMessage<RandomAccessFile>) exchange.getIn();
String fileName = inFileMessage.getGenericFile().getFileName();//文件名
String splitTag = File.separator;//系统文件分隔符
log.info(fileDir + splitTag + fileName);//文件的绝对路径
log.debug("业务代码处理");
//orderService.process(fileDir + splitTag + fileName);//解析入库等操作
}
}
三、application.properties 定义如下:
ftp.server.info=ftp://ftpadmin@192.168.1.220:21/dir?password=1qaz2wsx&delay=5s&move=done&readLock=rename
ftp.local.dir=file:D:/ftp/test
route.parserfile.info = {{ftp.local.dir}}?delay=10s&move=done&readLock=rename
route.parserfile.dir = {{ftp.local.dir}}/done
camel.springboot.main-run-controller=true
management.endpoint.camelroutes.enabled=true
management.endpoint.camelroutes.read-only=true
四、启动类
package com.lvyuanj.core.camel;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class CamelApplication {
public static void main(String[] args) {
SpringApplication.run(CamelApplication.class, args);
}
}
参数说明：
ftp.server.info=ftp://ftpadmin@192.168.1.220:21/dir?password=1qaz2wsx&delay=5s&move=done&readLock=rename
delay 采样时间频率；
move 表示下载文件完成后；
ftp端文件移动到当前目录下的done目录下
详细参数参考：http://camel.apache.org/ftp2.html
centos7 搭建FTP服务器
系统环境
Centos7
安装步骤
- 通过yum来安装
sudo yum -y install vsftpd
- 设置为开机启动
sudo chkconfig vsftpd on
- 创建ftp用户，比如ftpadmin。命令：
useradd -s /sbin/nologin -d /home/vsftpd ftpadmin
注意：
1）目录尽量不要选择根目录下，这里是/home/vsftpd，并且ftpadmin这个目录不要手动创建，否则权限会有问题，执行命令的时候会自动创建
2）注意目录的权限，如果有需要，应该设置相应的权限
4.为ftpadmin设置密码。命令：
passwd ftpadmin
执行之后输入2次密码确认就设置好了密码。
- 编辑vsftpd配置文件，命令:.
vim /etc/vsftpd/vsftpd.conf
找到anonymous_enable这个配置项，默认是YES，修改成NO，表示不允许匿名用户登录。
vsftpd的配置
vsftpd.ftpusers：位于/etc目录下。它指定了哪些用户账户不能访问FTP服务器，例如root等。
vsftpd.user_list：位于/etc目录下。该文件里的用户账户在默认情况下也不能访问FTP服务器，仅当vsftpd .conf配置文件里启用userlist_enable=NO选项时才允许访问。
vsftpd.conf：位于/etc/vsftpd目录下。来自定义用户登录控制、用户权限控制、超时设置、服务器功能选项、服务器性能选项、服务器响应消息等FTP服务器的配置。
/etc/vsftpd.conf 内容详解
#####################################
禁用匿名用户登陆
anonymous_enable=NO
允许本地用户登陆
local_enable=YES
允许本地用户写入
write_enable=YES
注意：这个地方如果不配置，就会出现只有root用户可以登陆，普通用户不可以
check_shell=NO
掩码，决定了上传上来的文件的权限。设置为000使之有最大权限
local_umask=000
允许记录日志
xferlog_enable=YES
允许数据流从20端口传输
connect_from_port_20=YES
日志路径
xferlog_file=/var/log/vsftpd.log
ftp欢迎语，可以随便设置
ftpd_banner=hi,guys!
注意：这个选项可以保证用户锁定在指定的家目录里，防止系统文件被修改。
chroot_local_user=YES
注意：这个不配置有可能出现只能下载不能上传
allow_writeable_chroot=YES
配置了可以以stand alone模式运行
listen=YES
说明我们要指定一个userlist，里边放的是允许ftp登陆的本地用户。如果设置为YES，则文件里设置的是不允许登陆的本地用户
userlist_deny=NO
userlist_enable=YES
记录允许本地登陆用户名的文件
userlist_file=/etc/vsftpd/allowed_users
#####################################
- 启动vsftp服务，命令：
systemctl start vsftpd.service
- 查看ftp服务的状态，命令：
systemctl status vsftpd.service
- 停止ftp服务，命令：
systemctl stop vsftpd.service
