Apache Camel继承Spring Boot 实现文件远程复制和转移

pom.xml

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-ftp</artifactId>
    <version>2.18.0</version>
</dependency>

URI的格式

ftp://[username@]hostname[:port]/directoryname[?options]

sftp://[username@]hostname[:port]/directoryname[?options]

ftps://[username@]hostname[:port]/directoryname[?options]

如果未提供端口号，Camel将根据协议提供默认值（ftp = 21，sftp = 22，ftps = 2222）。

例：
ftp://173.5.206.53:2121/MHE/?username=ftpAdmin&password=123456&binary=true&passiveMode=true&delete=true&delay=60000

注：ftp的路径必须是 当前ip:端口 + 当前账号密码登陆进去的目录往下走（不能往上走，没权限），路径才会生效，否则不会报错，也不会进行文件操作

配置文件信息：

#文件压缩本地地址(备份)
ftp.local.backups.dir=file:D:/gz/backups?move=delete&delay=30s
#大数据服务器地址(备份)
ftp.bigdata.backups.adress=sftp://218.984.130.146:51268/../apps/chinamobile/bigdata/dataConsistencyRepair/backups/?username=root&password=Redoor2018@net123&delay=30s


#大数据服务器修复地址（运维存放文件夹，使用后存放到备份文件中）
ftp.bigdatarepair.adress=sftp://218.984.130.146:51268/../apps/chinamobile/bigdata/bigdatarepair/?username=root&password=Redoor2018@net123&move=/apps/chinamobile/bigdata/bigdatarepair/backups&delay=30s
#本地文件修复地址（复制到自定义文件夹）
ftp.local.repair=file:D:/localrepair


#对比大数据差异文件存放地址（通知运维）
ftp.bigdatadifference.adress=sftp://218.984.130.146:51268/../apps/chinamobile/bigdata/difference/?username=root&password=Redoor2018@net123&delay=30s

camel.springboot.main-run-controller=true

文件实现文件复制，转移：

/**
 * 远程文件监听和拉取
 */
@Component
public class DownLoadRouteBuilder extends RouteBuilder{

    private Logger log = Logger.getLogger(DownLoadRouteBuilder.class);

    /**
     * 文件压缩本地地址
     */
    @Value("${ftp.local.dir}")
    private String localFileDir;

    /**
     * 大数据服务器地址
     */
    @Value("${ftp.bigdata.adress}")
    private String bigDataServer;

    /**
     * 文件压缩本地地址(上传备份)
     */
    @Value("${ftp.local.backups.dir}")
    private String localFileDirBackups;

    /**
     * 大数据服务器地址(上传备份)
     */
    @Value("${ftp.bigdata.backups.adress}")
    private String bigDataBackupsServer;

    /**
     * 本地文件修复地址
     */
    @Value("${ftp.local.repair}")
    private String localRepair;
    /**
     * 大数据服务器修复地址
     */
    @Value("${ftp.bigdatarepair.adress}")
    private String bigDataRepairServer;

    /**
     * 对比大数据差异文件存放地址
     */
    @Value("${ftp.bigdatadifference.adress}")
    private String bigdatadifference;

    @Override
    public void configure() throws Exception {
        //文件上传大数据
        from(localFileDir).to(bigDataServer).process("transferDataProcessToBigData");
        //文件上传大数据(上传备份)  备份文件不存redis有一份就好
        from(localFileDirBackups).to(bigDataBackupsServer).process("transferDataProcessToBigData");

        //修复文件拉取
        from(bigDataRepairServer).to(localRepair).process("transferDataProcessBigDataRepairToLocal");
        //对比大数据差异文件存放地址
        from(bigdatadifference).to(bigdatadifference).process("transferDataProcessToBigDataDifference");

        System.out.println("file download ok...");
        log.info("远程文件监听和拉取 已启动.......");
    }

}

文件复制获取信息：

　　　　文件远程复制和移动以后，本对象中可以显示

文件一：

/**
 * 监听本地修复文件
 */
@Component
public class TransferDataProcessBigDataRepairToLocal implements Processor{

    private Logger log = Logger.getLogger(TransferDataProcessBigDataRepairToLocal.class);

    @Autowired
    private RepairRedisUtil repairRedisUtil;
    /**
     * 本地文件修复地址
     */
    @Value("${ftp.local.repair}")
    private String localRepair;

    /**
     * redis工具类
     */
    @Autowired
    RedisUtils redisUtils;


    /**
     * 读取GZ压缩文件，多线程
     */
    @Autowired
    AsyncService asyncService;

    
    @Override
    public void process(Exchange exchange) throws Exception {
        GenericFileMessage<RandomAccessFile> inFileMessage = (GenericFileMessage<RandomAccessFile>) exchange.getIn();
        String fileName = inFileMessage.getGenericFile().getFileName();//文件名
        String splitTag = File.separator;//系统文件分隔符
        String absolutePath = localRepair + splitTag + fileName;
        System.out.println(absolutePath);//文件的绝对路径
        System.out.println("read file and push to falcon...");

        System.out.println("准备修复文件！");
        //文件的绝对路径去掉：file:
        String filePath = absolutePath.substring(absolutePath.indexOf("file:")+"file:".length());

        //创建redisKey  项目名称，模块名称，功能名称，自定义名称
        String bigdatarepair = redisUtils.buildRedisKey("OneLink","DataConsistencyRepair","bigdatarepair","list");

        //文件名称解析并进行redis缓存
        DataRepairCacheModel dataRepairModel = repairRedisUtil.getFileInfo(fileName);
        //获取缓存key
        String cacheKey = repairRedisUtil.getDataRepairCacheKey(dataRepairModel.getBusinessCode());
        //文件名称缓存
        repairRedisUtil.CacheRepairData(cacheKey,dataRepairModel,0);

    }
}

文件二：

/**
 * 监听大数据服务器是否上传成功
 */
@Component
public class TransferDataProcessToBigData implements Processor{

    private Logger log = Logger.getLogger(TransferDataProcessToBigData.class);

    /**
     * redis工具类
     */
    @Autowired
    RedisUtils redisUtils;

    /**
     * 大数据服务器地址
     */
    @Value("${ftp.bigdata.adress}")
    private String bigDataServer;

    
    @Override
    public void process(Exchange exchange) throws Exception {
        GenericFileMessage<RandomAccessFile> inFileMessage = (GenericFileMessage<RandomAccessFile>) exchange.getIn();
        String fileName = inFileMessage.getGenericFile().getFileName();//文件名
        String splitTag = File.separator;//系统文件分隔符
        System.out.println(bigDataServer + splitTag + fileName);//文件的绝对路径
        System.out.println("read file and push to falcon...");


        //添加：判断redis是否存在，不存在就删除，存在就不管

        //创建redisKey  项目名称，模块名称，功能名称，自定义名称
        String redisName = redisUtils.buildRedisKey("OneLink","DataConsistencyRepair","upload","List");
        //获取
        Map<String, DataUploadModel> stringObjectMap  = (Map<String, DataUploadModel>) JSON.parseObject((String)redisUtils.get(redisName),Map.class);

        //删除
        stringObjectMap.remove(fileName);

        //将集合存入缓存中
        redisUtils.set(redisName, JSONObject.toJSONString(stringObjectMap));

        log.info("文件信息："+fileName+"上传大数据服务器成功！");
        log.info("文件信息："+fileName+"从redis删除标记！");
    }
}

文件三：

/**
 * 对比大数据差异文件存放地址监听
 */
@Component
public class TransferDataProcessToBigDataDifference implements Processor{

    private Logger log = Logger.getLogger(TransferDataProcessToBigDataDifference.class);

    /**
     * redis工具类
     */
    @Autowired
    RedisUtils redisUtils;

    @Autowired
    private Task task;

    /**
     * 对比大数据差异文件存放地址
     */
    @Value("${ftp.bigdatadifference.adress}")
    private String bigdatadifference;

    @Override
    public void process(Exchange exchange) throws Exception {
        GenericFileMessage<RandomAccessFile> inFileMessage = (GenericFileMessage<RandomAccessFile>) exchange.getIn();
        String fileName = inFileMessage.getGenericFile().getFileName();//文件名
        String splitTag = File.separator;//系统文件分隔符
        System.out.println(bigdatadifference + splitTag + fileName);//文件的绝对路径
        System.out.println("read file and push to falcon...");
        task.send();
        System.out.println("给运维发送邮件！");
    }
}

官方API文档：http://camel.apache.org/ftp2.html