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

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

1. 依赖

<dependency>
    <groupId>org.apache.camel.springboot</groupId>
    <artifactId>camel-spring-boot-starter</artifactId>
    <version>3.4.2</version>
</dependency>
<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-ftp</artifactId>
    <version>3.4.2</version>
</dependency>
版本保持一致,可用最新版。

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/

2.1 配置文档#

名称 说明 默认值
username 登录用户名
password 登录密码
binary 指定文件传输模式,BINARY 或 ASCII false(ASCII)
charset 指定文件读取的编码格式,写出时也可配置
disconnect 使用后是否立即断开连接 false
passiveMode 是否启用被动模式 false
separator 设置路径分隔符,可选:UNIX,Windows,Auto UNIX
delete 是否在文件处理完成后删除源文件 false
noop 如果 noop = true,Camel 也将设置 idempotent = true 以避免一遍又一遍地使用相同的文件 false
recursive 是否遍历文件夹处理文件 false
download 是否下载文件,如果将此选项设置为 false,则消息正文将为 null
filter 设置过滤器
idempotent 选择使用幂等消费者 EIP 模式让 Camel 跳过已处理的文件。如果 noop = true,则将启用幂等,以避免重复使用相同的文件。
delay 间隔多少 ms 轮询一次
reconnectDelay 延迟 n 毫秒,然后再执行重新连接尝试

3. 编写一个Component

继承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;
/**
 * @className: DownloadRouteDemo
 * @description: DownloadRouteDemo
 */
@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); //处理器
    }
}

4. 过滤器设置

过滤器设置,只处理.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;
/**
 * @className: FtpDownloadFileFilter
 * @description: FtpDownloadFileFilter
 */
@Component
public class FtpDownloadFileFilter implements Predicate {
    private static Logger logger = LoggerFactory.getLogger( FtpDownloadFileFilter.class );
    @Override
    public boolean matches(Exchange exchange) {
        boolean bMatches = true;
        GenericFileMessage<RandomAccessFile> inFileMessage = (GenericFileMessage<RandomAccessFile>) exchange.getIn();
        String fileName = inFileMessage.getGenericFile().getFileName();
        bMatches = fileName.endsWith(".txt") || fileName.endsWith(".json");
        logger.info("fileName:"+fileName+"matches:"+bMatches);
        return bMatches;
    }
}

5. 处理器

代码如下:

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;
/**
 * @className: LocationFileProcessor
 * @description: LocationFileProcessor
 */
@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<RandomAccessFile> inFileMessage = (GenericFileMessage<RandomAccessFile>) exchange.getIn();
        //文件名
        String fileName = inFileMessage.getGenericFile().getFileName();
        //文件的绝对路径
        logger.info(fileDir + fileName);
        //解析入库等操作
        orderService.process(fileDir + fileName);
    }
}

6. 定义一个OrderService

package com.example.goodsync.ftpdemo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
/**
 * @className: OrderService
 * @description: OrderService
 */
@Service
public class OrderService {
    private static Logger logger = LoggerFactory.getLogger( OrderService.class );
    public void process(String path){
        logger.info("path:"+path);
    }
}

7. 问题

7.1 更多的文档信息移步官方文档#

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

7.2 出现“550 create directory operation failed”错误#

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

解决办法: 关闭SELinux

# vim /etc/selinux/config

将 SELINUX=XXX -->XXX 代表级别 改为 SELINUX=disabled

重启系统

7.3 文件名出现乱码#

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

相关文章
|
1月前
|
数据可视化 Java API
Spring Boot与Swagger的集成
Spring Boot与Swagger的集成
|
1月前
|
Java API 开发者
在Spring Boot中集成Swagger API文档
在Spring Boot中集成Swagger API文档
|
4天前
|
NoSQL 关系型数据库 MySQL
SpringBoot 集成 SpringSecurity + MySQL + JWT 附源码,废话不多直接盘
SpringBoot 集成 SpringSecurity + MySQL + JWT 附源码,废话不多直接盘
13 2
|
16天前
|
NoSQL Java Redis
Spring Boot集成Redis全攻略:高效数据存取,打造性能飞跃的Java微服务应用!
【8月更文挑战第3天】Spring Boot是备受欢迎的微服务框架,以其快速开发与轻量特性著称。结合高性能键值数据库Redis,可显著增强应用性能。集成步骤包括:添加`spring-boot-starter-data-redis`依赖,配置Redis服务器参数,注入`RedisTemplate`或`StringRedisTemplate`进行数据操作。这种集成方案适用于缓存、高并发等场景,有效提升数据处理效率。
70 2
|
22天前
|
监控 druid Java
spring boot 集成配置阿里 Druid监控配置
spring boot 集成配置阿里 Druid监控配置
120 6
|
20天前
|
JSON 缓存 Java
Spring Boot集成 Swagger2 展现在线接口文档
本节课详细分析了 Swagger 的优点,以及 Spring Boot 如何集成 Swagger2,包括配置,相关注解的讲解,涉及到了实体类和接口类,以及如何使用。最后通过页面测试,体验了 Swagger 的强大之处,基本上是每个项目组中必备的工具之一,所以要掌握该工具的使用,也不难。
|
1月前
|
Java 关系型数据库 MySQL
如何实现Springboot+camunda+mysql的集成
【7月更文挑战第2天】集成Spring Boot、Camunda和MySQL的简要步骤: 1. 初始化Spring Boot项目,添加Camunda和MySQL驱动依赖。 2. 配置`application.properties`,包括数据库URL、用户名和密码。 3. 设置Camunda引擎属性,指定数据源。 4. 引入流程定义文件(如`.bpmn`)。 5. 创建服务处理流程操作,创建控制器接收请求。 6. Camunda自动在数据库创建表结构。 7. 启动应用,测试流程启动,如通过服务和控制器开始流程实例。 示例代码包括服务类启动流程实例及控制器接口。实际集成需按业务需求调整。
139 4
|
1月前
|
SQL Java 数据库
实时计算 Flink版产品使用问题之Spring Boot集成Flink可以通过什么方式实现通过接口启动和关闭Flink程序
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
1月前
|
Java API 开发工具
Spring Boot与Spring Cloud Config的集成
Spring Boot与Spring Cloud Config的集成
|
1月前
|
开发框架 Java 数据库
Spring Boot集成多数据源的最佳实践
Spring Boot集成多数据源的最佳实践

推荐镜像

更多