Apache Camel 与 Spring Boot 集成，通过FTP定时采集、处理文件

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

安装步骤
  1. 通过yum来安装
sudo yum -y install vsftpd
  1. 设置为开机启动
sudo chkconfig vsftpd on
  1. 创建ftp用户，比如ftpadmin。命令：
    useradd -s /sbin/nologin -d /home/vsftpd ftpadmin

注意：
1）目录尽量不要选择根目录下，这里是/home/vsftpd，并且ftpadmin这个目录不要手动创建，否则权限会有问题，执行命令的时候会自动创建
2）注意目录的权限，如果有需要，应该设置相应的权限

4.为ftpadmin设置密码。命令：

passwd ftpadmin

执行之后输入2次密码确认就设置好了密码。

  1. 编辑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

#####################################

  1. 启动vsftp服务，命令：
systemctl start vsftpd.service
  1. 查看ftp服务的状态，命令：
systemctl status vsftpd.service
  1. 停止ftp服务，命令：
systemctl stop vsftpd.service
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容