Apache Camel 與 Spring Boot 集成，通過(guò)FTP定時(shí)采集、處理文件

Apache Camel 與 Spring Boot 集成，通過(guò)FTP定時(shí)采集、處理文件

一、maven項(xiàng)目需要引入相關(guān)包

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 相關(guān)包 -->
        <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 內(nèi)部使用切面代理，不引入會(huì)報(bào)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>

二、編寫一個(gè)Component，擴(kuò)展RouteBuilder.class,重新實(shí)現(xiàn)configure方法。設(shè)置數(shù)據(jù)源、目標(biāo)，并且可以根據(jù)需要額外設(shè)置 過(guò)濾器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;//系統(tǒng)文件分隔符
        log.info(fileDir + splitTag + fileName);//文件的絕對(duì)路徑
        log.debug("業(yè)務(wù)代碼處理");
        //orderService.process(fileDir + splitTag + fileName);//解析入庫(kù)等操作
    }

}

三、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

四、啟動(dòng)類

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);
    }
}
參數(shù)說(shuō)明：
ftp.server.info=ftp://ftpadmin@192.168.1.220:21/dir?password=1qaz2wsx&delay=5s&move=done&readLock=rename

delay 采樣時(shí)間頻率；
move 表示下載文件完成后；
ftp端文件移動(dòng)到當(dāng)前目錄下的done目錄下
詳細(xì)參數(shù)參考：http://camel.apache.org/ftp2.html

centos7 搭建FTP服務(wù)器

系統(tǒng)環(huán)境

Centos7

安裝步驟
  1. 通過(guò)yum來(lái)安裝
sudo yum -y install vsftpd
  1. 設(shè)置為開機(jī)啟動(dòng)
sudo chkconfig vsftpd on
  1. 創(chuàng)建ftp用戶，比如ftpadmin。命令：
    useradd -s /sbin/nologin -d /home/vsftpd ftpadmin

注意：
1）目錄盡量不要選擇根目錄下，這里是/home/vsftpd，并且ftpadmin這個(gè)目錄不要手動(dòng)創(chuàng)建，否則權(quán)限會(huì)有問(wèn)題，執(zhí)行命令的時(shí)候會(huì)自動(dòng)創(chuàng)建
2）注意目錄的權(quán)限，如果有需要，應(yīng)該設(shè)置相應(yīng)的權(quán)限

4.為ftpadmin設(shè)置密碼。命令：

passwd ftpadmin

執(zhí)行之后輸入2次密碼確認(rèn)就設(shè)置好了密碼。

  1. 編輯vsftpd配置文件，命令:.
vim /etc/vsftpd/vsftpd.conf

找到anonymous_enable這個(gè)配置項(xiàng)，默認(rèn)是YES，修改成NO，表示不允許匿名用戶登錄。
vsftpd的配置

vsftpd.ftpusers：位于/etc目錄下。它指定了哪些用戶賬戶不能訪問(wèn)FTP服務(wù)器，例如root等。

vsftpd.user_list：位于/etc目錄下。該文件里的用戶賬戶在默認(rèn)情況下也不能訪問(wèn)FTP服務(wù)器，僅當(dāng)vsftpd .conf配置文件里啟用userlist_enable=NO選項(xiàng)時(shí)才允許訪問(wèn)。

vsftpd.conf：位于/etc/vsftpd目錄下。來(lái)自定義用戶登錄控制、用戶權(quán)限控制、超時(shí)設(shè)置、服務(wù)器功能選項(xiàng)、服務(wù)器性能選項(xiàng)、服務(wù)器響應(yīng)消息等FTP服務(wù)器的配置。

/etc/vsftpd.conf 內(nèi)容詳解

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

禁用匿名用戶登陸

anonymous_enable=NO

允許本地用戶登陸

local_enable=YES

允許本地用戶寫入

write_enable=YES

注意：這個(gè)地方如果不配置，就會(huì)出現(xiàn)只有root用戶可以登陸，普通用戶不可以

check_shell=NO

掩碼，決定了上傳上來(lái)的文件的權(quán)限。設(shè)置為000使之有最大權(quán)限

local_umask=000

允許記錄日志

xferlog_enable=YES

允許數(shù)據(jù)流從20端口傳輸

connect_from_port_20=YES

日志路徑

xferlog_file=/var/log/vsftpd.log

ftp歡迎語(yǔ)，可以隨便設(shè)置

ftpd_banner=hi,guys!

注意：這個(gè)選項(xiàng)可以保證用戶鎖定在指定的家目錄里，防止系統(tǒng)文件被修改。

chroot_local_user=YES

注意：這個(gè)不配置有可能出現(xiàn)只能下載不能上傳

allow_writeable_chroot=YES

配置了可以以stand alone模式運(yùn)行

listen=YES

說(shuō)明我們要指定一個(gè)userlist，里邊放的是允許ftp登陸的本地用戶。如果設(shè)置為YES，則文件里設(shè)置的是不允許登陸的本地用戶

userlist_deny=NO

userlist_enable=YES

記錄允許本地登陸用戶名的文件

userlist_file=/etc/vsftpd/allowed_users

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

  1. 啟動(dòng)vsftp服務(wù)，命令：
systemctl start vsftpd.service
  1. 查看ftp服務(wù)的狀態(tài)，命令：
systemctl status vsftpd.service
  1. 停止ftp服務(wù)，命令：
systemctl stop vsftpd.service
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容