SpringBoot 集成 Apache Camel FTP 实现文件同步

    org.apache.camel.springboot
    camel-spring-boot-starter
    3.4.2


    org.apache.camel
    camel-ftp
    3.4.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/

继承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;
 

@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); //处理器
    }
}

过滤器设置，只处理.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;
 
 

 
@Component
public class FtpDownloadFileFilter implements Predicate {
    private static Logger logger = LoggerFactory.getLogger( FtpDownloadFileFilter.class );
 
    @Override
    public boolean matches(Exchange exchange) {
        boolean bMatches = true;
        GenericFileMessage inFileMessage = (GenericFileMessage) exchange.getIn();
        String fileName = inFileMessage.getGenericFile().getFileName();
        bMatches = fileName.endsWith(".txt") || fileName.endsWith(".json");
 
        logger.info("fileName:"+fileName+"matches:"+bMatches);
 
        return bMatches;
    }
}

代码如下:

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;


@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 inFileMessage = (GenericFileMessage) exchange.getIn();

        //文件名
        String fileName = inFileMessage.getGenericFile().getFileName();
        //文件的绝对路径
        logger.info(fileDir + fileName);
        //解析入库等操作
        orderService.process(fileDir + fileName);
 
    }
 
}

package com.example.goodsync.ftpdemo;
 
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
 

@Service
public class OrderService {
    private static Logger logger = LoggerFactory.getLogger( OrderService.class );
    public void process(String path){
        logger.info("path:"+path);
    }
}

https://camel.apache.org/components/3.14.x/ftp-component.html#_endpoint_query_option_transferLoggingVerbose

原因是SELinux(Security-Enhanced Linux----是美国国家安全局对于强制访问控制的实现, 是Linux 上最杰出的新安全子系统.)安装机制搞的鬼.只要disable SELinux就可以了.
解决办法: 关闭SELinux

# vim /etc/selinux/config

将 SELINUX=XXX -->XXX 代表级别 改为 SELINUX=disabled
重启系统

地址后面拼接 ftpClient.controlEncoding：配置 ftpClient 编码格式。 比如ftpClient.controlEncoding=UTF-8

备注：此文章从本人博客园转移而来，本人博客园地址：https://www.cnblogs.com/equals