本篇博文介绍spring integration sftp技术中的sftp outbound gateway相关内容。Sftp outbound gateway 其实质就是提供一组命令（如图1）来实现对服务器上文件的交互操作，包括文件的获取（文件对象和文件名等）、上传（单文件和多文件）、下载（单文件和多文件），删除，移动。具体在开发的过程中可以使用多种配置方式如xml，springboot等。本文在介绍SFTP Outbound Gateway 的基础上，使用SpringBoot开发框架进行相应的开发实践。
1.命令组
1.1 ls
该命令的功能是获取远程文件，包括文件对象和文件路径名称等，具体返回值根据配置的选项：
- -1 ：获取一组远程文件的文件名；默认是获取一组FileInfo对象；
- -a：获取所有的文件（包括开始的文件，递归时使用）；
- - f：检索结果不用排序；
- -dirs： 包括文件夹，默认是包括的；
- -links：包括链接符号，默认是包括的；
- -R：递归方式获取远程文件夹下所有文件，默认不递归的。
除此之外，还可以配置文件名过滤器等；
命令返回值： 通过ls命令获取的message payload，是一组文件名或者FileInfo对象，对象中提供了有关文件的修改时间，权限以及其他的信息；
ls命令作用的远程文件夹，由header头的file_remoteDirectory属性提供；
建议提醒：如果使用-R递归选择项，文件名将含有子文件夹，表明递归文件的相对路径；如果使用-dirs选项，每一个递归的子文件夹，返回的元素中将含有子文件夹名；在这种情况下，建议不用使用-1罗列文件名，因为返回的元素中不能够区分是文件还是文件夹？建议返回FileInfo对象。
下面是开发示例：
@Bean
@ServiceActivator(inputChannel = "sftpChannel2")
public MessageHandler handler2() {
//指定session配置和命令
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpSessionFactory(),"ls","payload");
sftpOutboundGateway.setOptions("-dirs"); //配置项
return sftpOutboundGateway;
}
//使用Gateway触发
@MessagingGateway
public interface MessageGateway {
@Gateway(requestChannel = "sftpChannel2")
List<FileInfo> listFileName(String dir); //指定远程文件夹
}
1.2 nlst
该命令提供检索远程文件名的功能，相当于ls -1的命令；支持如下配置：
- -f：文件名不排序；
nlst命令作用的远程文件夹，由header头的file_remoteDirectory提供。
返回值：通过nlst获取的文件message payload，就是一组文件名列表；
1.3 get
该命令由于获取一个远程的文件，支持如下的选项：
- -P：文件下载之后，保持文件在本地的时间戳同远程服务器一致；
- -stream：以流的方式获取远程文件；
- -D：文件成功转移之后，删除远程文件；如果FileExistsMode设置为IGNORE，远程文件不会删除。
file_remoteDirectory 头包含了文件的远程路径，file_remoteFile属性为文件名；
返回值：使用get方法获取的message的payload是一个File对象，如果使用-straem，则payload就是一个InputStream文件流。
对于文本文件，有个通用的案例，使用file splitter 或 stream transformer。当以文件流的形式获取远程文件，Session在结束之后要及时关闭. Session由closeableResource属性header头文件，IntegrationMessageHeaderAccessor提供了流资源的关闭操作。
1.4 mget
该命令用来基于特定的文件模式过滤器获取多个文件，支持如下的设置：
- -P：保留远程文件的时间戳；
- -R：递归下载所有符合的文件；
- -x：没有文件匹配文件筛选模式，抛出异常，并返回空集合；
- -D：文件成功转移之后。如何FileExistsMode=IGNORE，本地文件存在，文件不会删除；
message payload返回的是List< >对象,集合元素是File。
注意：
在5.0版本之后，若FileExistsMode=IGNORE，payload不再包含已经存在的文件对象。
remote path的表达式应该是以结尾，类似myfiles/,表示获取完整的文件夹树myfiles；
注意，在版本5.0之后，MGET命令可以设置FileExistsMode.REPLACE_IF_MODIFIED模式，去同步整个文件夹，被修改的文件的时间戳也会相应修改。不用关心-P模式；
-R模式，默认情况下是整个文件夹，同时也支持设置文件或文件夹过滤器FileListFilter； 该过滤器提供两种方式filename-pattern或者filename-regex属性；例如filename-regex="(subDir|.*1.txt)" 获取subDir下所有以1.txt结尾的文件；
通常，将在local-directory-expression中使用#remoteDirectory变量，以便远程目录结构在本地保留。
下面是开发示例：
@Bean
@ServiceActivator(inputChannel = "sftpChannel3")
public MessageHandler handler3() {
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpSessionFactory(),"mget","payload");
sftpOutboundGateway.setOptions("-R");
sftpOutboundGateway.setFileExistsMode(FileExistsMode.REPLACE_IF_MODIFIED);
sftpOutboundGateway.setLocalDirectory(new File("E:\sftp_tmp_dir"));
sftpOutboundGateway.setAutoCreateLocalDirectory(true);
return sftpOutboundGateway;
}
@MessagingGateway
public interface MessageGateway {
@Gateway(requestChannel = "sftpChannel3")
List<File> listFile(String dir);
}
1.5 put
该命令是发送单个文件到远程服务器；
message的payload可以是File对象，byte[]数组，或者字符串；
remote-filename-generator用来命名远程文件。其他的属性如remote-directory，temporary-remote-directory等等；
返回值：put命令的message的payload的返回值是string，包含文件传输后在服务器上的整个路径；
1.6 mput
该命令是发送多个文件到服务器，支持如下配置：
- -R: 递归发送文件和子文件夹下的所有文件；
message payload必须是文件或者文件路径字符串，代表了本地文件夹；自版本5.1之后，也支持文件或者路径字符串集合；
put的配置，同样适合mput，同时除此之外，还提供过滤文件的mput-pattern，mput-regex，mput-filter等；
版本4.3之后，支持设置文件的权限；
返回值：mput执行之后的返回值，是一个List，包含文件转移之后的路径集合。
下面是开发示例：
//!important，put命令需要借助与sftpRemoteFileTemplate。
//看源码，可以发现outbound gateway 有多种构造函数；
@Bean
@ServiceActivator(inputChannel = "sftpChannel4")
public MessageHandler handler4(){
SftpRemoteFileTemplate sftpRemoteFileTemplate = new SftpRemoteFileTemplate(sftpSessionFactory());
sftpRemoteFileTemplate.setRemoteDirectoryExpression(new LiteralExpression("/send"));
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpRemoteFileTemplate,"put","payload");
sftpOutboundGateway.setBeanFactory(beanFactory);
return sftpOutboundGateway;
}
@Bean
@ServiceActivator(inputChannel = "sftpChannel5")
public MessageHandler handler5(){
SftpRemoteFileTemplate sftpRemoteFileTemplate = new SftpRemoteFileTemplate(sftpSessionFactory());
sftpRemoteFileTemplate.setRemoteDirectoryExpression(new LiteralExpression("/send"));
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpRemoteFileTemplate,"mput","payload");
//配置过滤器
sftpOutboundGateway.setMputFilter(new FileListFilter<File>() {
@Override
public List<File> filterFiles(File[] files) {
if(...){
...
}
return null;
}
});
sftpOutboundGateway.setBeanFactory(beanFactory);
return sftpOutboundGateway;
}
1.7 rm
该命令是删除远程文件。
如果删除成功，message payload的返回值是Boolean.TRUE；否则是Boolean.FALSE。
file_remoteDirectory头包含远程文件属性；
下面是开发示例：
@Bean
@ServiceActivator(inputChannel = "sftpChannel6")
public MessageHandler handler6(){
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpSessionFactory(),"rm","payload");
sftpOutboundGateway.setBeanFactory(beanFactory);
return sftpOutboundGateway;
}
1.8 mv
该命令是移动文件在远程服务器上的位置。
返回值：转移成功，返回true，否则是false；
下面是开发示例：
@Bean
@ServiceActivator(inputChannel = "sftpChannel7")
public MessageHandler handler7(){
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpSessionFactory(),"mv","'send/22.TXT'");
sftpOutboundGateway.setRenameExpression(new LiteralExpression("send1/22.TXT"));
sftpOutboundGateway.setBeanFactory(beanFactory);
return sftpOutboundGateway;
}
以下是干货（测试用例）：
首先是POM文件：
<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.flower.springintegration</groupId>
<artifactId>spring-integration-samples</artifactId>
<version>v0.0.1</version>
<name>SpringIntegrationExamples</name>
<description>Spring Integration Samples</description>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
</properties>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.1.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- <dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-batch</artifactId>
</dependency>-->
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-sftp</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/org.quartz-scheduler/quartz -->
<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>2.2.1</version>
</dependency>
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-test</artifactId>
<version>1.4.0.RELEASE</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>4.3.2.RELEASE</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
接下来是yml文件配置：
spring:
datasource:
type: com.zaxxer.hikari.HikariDataSource
url: jdbc:mysql://localhost:3306/springbatchexample?serverTimezone=GMT%2B8&useUnicode=true&characterEncoding=utf-8
username: root
password: root
sftp:
host: 127.0.0.1
port: 23
user: 47gamer
password: wdnmd
filePath:
send: /send
achieve: /achieve
localPath: /sftp_tmp_dir
然后是Sftp网关配置类SftpConfig.java
package com.flower.integration.sftp;
import com.jcraft.jsch.ChannelSftp.LsEntry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.annotation.*;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.file.remote.FileInfo;
import org.springframework.integration.file.remote.session.CachingSessionFactory;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.file.support.FileExistsMode;
import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
import org.springframework.integration.sftp.gateway.SftpOutboundGateway;
import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer;
import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizingMessageSource;
import org.springframework.integration.sftp.outbound.SftpMessageHandler;
import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import javax.annotation.Resource;
import java.io.File;
import java.util.List;
import java.util.Properties;
/**
* Sftp configuration.
*
* @Autor 47Gamer
* @Date 2019-01-18
*/
@Configuration
@DependsOn("sftpProperty")
public class SftpConfig {
@Resource(name = "sftpProperty")
private SftpProperty sftpProperty;
private static Logger log = LoggerFactory.getLogger(SftpConfig.class);
@Value("${sftp.host}")
private String sftpHost;
@Value("${sftp.port:23}")
private int sftpPort;
@Value("${sftp.user}")
private String sftpUser;
@Value("${sftp.privateKey:#{null}}")
private org.springframework.core.io.Resource sftpPrivateKey;
@Value("${sftp.privateKeyPassphrase:}")
private String sftpPrivateKeyPassphrase;
@Value("${sftp.password}")
private String sftpPassword;
/* @Bean
public SessionFactory<LsEntry> sftpSessionFactory() {
System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(false);
factory.setHost(sftpProperty.getHost());
factory.setPort(sftpProperty.getPort());
factory.setUser(sftpProperty.getUser());
Properties jschProps = new Properties();
//!important 必须配置PreferredAuthentications，否则程序控制台会询问user name 和 password。
jschProps.put("StrictHostKeyChecking", "no");
jschProps.put("PreferredAuthentications",
"password,gssapi-with-mic,publickey,keyboard-interactive");
factory.setSessionConfig(jschProps);
// if (sftpPassword != null) {
factory.setPassword(sftpProperty.getPassword());
// } else {
// factory.setPrivateKey(sftpPrivateKey);
// factory.setPrivateKeyPassphrase(sftpPrivateKeyPassphrase);
// }
factory.setAllowUnknownKeys(true);
// //设置缓存的属性，缓存的size(), waitTimeout().
CachingSessionFactory<LsEntry> cachingSessionFactory =
new CachingSessionFactory<LsEntry>(factory);
cachingSessionFactory.setPoolSize(10);
// cachingSessionFactory.setSessionWaitTimeout(1000);
return cachingSessionFactory;
// return new CachingSessionFactory<LsEntry>(factory);
}*/
/**
* 创建 spring-integration-sftp session
* 避免使用jsch原生的创建session的方式
*
* @return SessionFactory<LsEntry>
*/
@Bean
public SessionFactory<LsEntry> sftpSessionFactory(){
System.out.println("######################################################");
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setUser(sftpProperty.getUser());
factory.setHost(sftpProperty.getHost());
factory.setPort(sftpProperty.getPort());
factory.setPassword(sftpProperty.getPassword());
Properties jschProps = new Properties();
//!important 必须配置PreferredAuthentications，否则程序控制台会询问user name 和 password。
jschProps.put("StrictHostKeyChecking", "no");
jschProps.put("PreferredAuthentications",
"password,gssapi-with-mic,publickey,keyboard-interactive");
factory.setSessionConfig(jschProps);
factory.setAllowUnknownKeys(true);
//设置缓存的属性，缓存的size(), waitTimeout().
CachingSessionFactory<LsEntry> cachingSessionFactory =
new CachingSessionFactory<LsEntry>(factory);
// cachingSessionFactory.setPoolSize(2000);
return cachingSessionFactory;
}
/**
* 配置Outbound Channel Adapter.
*
* 实质上就是一个MessageHandler，接收Message Channel发送的信息流.
* @return MessageHandler
*/
@ServiceActivator(inputChannel = "fileInChannel")
@Bean
public SftpMessageHandler sftpMessageHandler(){
SftpMessageHandler sftpMsgHandler = new SftpMessageHandler(sftpSessionFactory());
sftpMsgHandler.setRemoteDirectoryExpression(
new LiteralExpression(sftpProperty.getSftpAchievePath()));
sftpMsgHandler.setAutoCreateDirectory(true);
sftpMsgHandler.setCharset("UFT-8");
return sftpMsgHandler;
}
/**
* 配置 Inbound Channel Adapter
*
* 监控sftp服务器文件的状态。一旦由符合条件的文件生成，就将其同步到本地服务器。
* 需要条件：inboundFileChannel的bean；轮询的机制；文件同步bean,SftpInboundFileSynchronizer；
*/
@Bean
@InboundChannelAdapter(value = "inboundFileChannel",
poller = @Poller(cron = "0 1/10 * * * *", maxMessagesPerPoll = "1"))
public MessageSource<File> fileMessageSource() {
System.out.println("=========================================================");
//创建sftpInboundFileSynchronizer，并绑定到message source.
SftpInboundFileSynchronizingMessageSource source =
new SftpInboundFileSynchronizingMessageSource(sftpInboundFileSynchronizer());
//自动创建本地文件夹
source.setAutoCreateLocalDirectory(true);
source.setLocalDirectory(new File(sftpProperty.getLocalTempDir()));
//设置文件过滤器
source.setLocalFilter(new AcceptOnceFileListFilter<File>());
return source;
}
/**
* 为Inbound-channel-adapter提供bean
*/
@Bean
public DirectChannel inboundFileChannel() {
return new DirectChannel();
}
/**
* SftpInboundFileSynchronizer,
*
* 同步sftp文件至本地服务器.
* <1> 可以放在service中获取bean使用.toLocal方法；
* <2> 也可以使用inbound-channel-adapter中，做监控文件服务器的动态。
*
* @return SftpInboundFileSynchronizer
*/
@Bean(name = "synFileChannel")
public SftpInboundFileSynchronizer sftpInboundFileSynchronizer (){
SftpInboundFileSynchronizer fileSynchronize =
new SftpInboundFileSynchronizer(sftpSessionFactory());
fileSynchronize.setDeleteRemoteFiles(true);
fileSynchronize.setPreserveTimestamp(true);
//!important
fileSynchronize.setRemoteDirectory(sftpProperty.getSftpSendPath());
fileSynchronize.setFilter(new SftpSimplePatternFileListFilter("*.*"));
//fileSynchronize.setLocalFilenameGeneratorExpression( );
fileSynchronize.setPreserveTimestamp(true);
return fileSynchronize;
}
///////////////////////////////////////////////////////////////////////
/**
* 配置 SFTP Outbound Gateway
*
* @return MessageHandler
*/
@Bean
@ServiceActivator(inputChannel = "sftpChannel")
public MessageHandler handler() {
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpSessionFactory(),"ls","payload");
// MessageChannel message = sftpOutboundGateway.getOutputChannel();
sftpOutboundGateway.setLocalDirectory(new File("E:\sftp_tmp_dir"));
sftpOutboundGateway.setAutoCreateLocalDirectory(true); // TODO dynanic path
return sftpOutboundGateway;
}
@Bean
@ServiceActivator(inputChannel = "sftpChannel2")
public MessageHandler handler2() {
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpSessionFactory(),"ls","payload");
sftpOutboundGateway.setOptions("-dirs");
sftpOutboundGateway.setLocalDirectory(new File("E:\sftp_tmp_dir"));
sftpOutboundGateway.setAutoCreateLocalDirectory(true); // TODO dynanic path
return sftpOutboundGateway;
}
@Bean
@ServiceActivator(inputChannel = "sftpChannel3")
public MessageHandler handler3() {
System.out.println("========================= 3 ================================");
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpSessionFactory(),"mget","payload");
sftpOutboundGateway.setOptions("-R");
sftpOutboundGateway.setFileExistsMode(FileExistsMode.REPLACE_IF_MODIFIED);
sftpOutboundGateway.setLocalDirectory(new File("E:\sftp_tmp_dir"));
sftpOutboundGateway.setAutoCreateLocalDirectory(true); // TODO dynanic path
return sftpOutboundGateway;
}
@Autowired
private BeanFactory beanFactory;
//outbound gateway，put命令需要借助与sftpRemoteFileTemplate。
//看源码，可以发现outbound gateway 有多种构造函数；
@Bean
@ServiceActivator(inputChannel = "sftpChannel4")
public MessageHandler handler4(){
SftpRemoteFileTemplate sftpRemoteFileTemplate = new SftpRemoteFileTemplate(sftpSessionFactory());
sftpRemoteFileTemplate.setRemoteDirectoryExpression(new LiteralExpression("/send"));
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpRemoteFileTemplate,"put","payload");
// sftpOutboundGateway.setLocalDirectoryExpressionString("/get/");
sftpOutboundGateway.setBeanFactory(beanFactory);
return sftpOutboundGateway;
}
@Bean
@ServiceActivator(inputChannel = "sftpChannel5")
public MessageHandler handler5(){
SftpRemoteFileTemplate sftpRemoteFileTemplate = new SftpRemoteFileTemplate(sftpSessionFactory());
sftpRemoteFileTemplate.setRemoteDirectoryExpression(new LiteralExpression("/send"));
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpRemoteFileTemplate,"mput","payload");
// sftpOutboundGateway.setLocalDirectoryExpressionString("/get/");
// sftpOutboundGateway.setOptions("-R");
sftpOutboundGateway.setMputFilter(new FileListFilter<File>() {
@Override
public List<File> filterFiles(File[] files) {
return null;
}
});
sftpOutboundGateway.setBeanFactory(beanFactory);
return sftpOutboundGateway;
}
@Bean
@ServiceActivator(inputChannel = "sftpChannel6")
public MessageHandler handler6(){
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpSessionFactory(),"rm","payload");
sftpOutboundGateway.setBeanFactory(beanFactory);
return sftpOutboundGateway;
}
@Bean
@ServiceActivator(inputChannel = "sftpChannel7")
public MessageHandler handler7(){
// SftpRemoteFileTemplate sftpRemoteFileTemplate = new SftpRemoteFileTemplate(sftpSessionFactory());
// sftpRemoteFileTemplate.setRemoteDirectoryExpression(new LiteralExpression("/send"));
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpSessionFactory(),"mv","'send/22.TXT'");
// sftpOutboundGateway.setRenameExpression(new LiteralExpression("/send1"));
// sftpOutboundGateway.setChmod(777);
// sftpOutboundGateway.setRenameExpressionString("send1");
sftpOutboundGateway.setRenameExpression(new LiteralExpression("send1/22.TXT"));
// sftpOutboundGateway.setAutoCreateLocalDirectory(true);
sftpOutboundGateway.setBeanFactory(beanFactory);
return sftpOutboundGateway;
}
@MessagingGateway
public interface UploadGateway {
@Gateway(requestChannel = "sftpChannel")
List<FileInfo> listFileInfo(String dir);
@Gateway(requestChannel = "sftpChannel2")
List<FileInfo> listFileName(String dir);
@Gateway(requestChannel = "sftpChannel3")
List<File> listFile(String dir);
@Gateway(requestChannel = "sftpChannel4")
String putFile(File source);
@Gateway(requestChannel = "sftpChannel5")
List<String> mputFile(File directory);
@Gateway(requestChannel = "sftpChannel6")
boolean removeFile(String file);
@Gateway(requestChannel = "sftpChannel7")
boolean moveFile(String file);
}
}
映射yml文件里的stfp配置实体SftpProperty .java
package com.flower.integration.sftp;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
import java.util.Map;
@Component("sftpProperty")
@ConfigurationProperties(prefix = "sftp")
public class SftpProperty {
private String host;
private Integer port;
private String user;
private String password;
private Map<String,String> filePath;
////////////////////////////////////////////////////
public String getSftpSendPath(){
return filePath.get("send");
}
public String getSftpAchievePath(){
return filePath.get("achieve");
}
public String getLocalTempDir(){
return filePath.get("localPath");
}
///////////////////////////////////////////////////
public String getHost() {
return host;
}
public void setHost(String host) {
this.host = host;
}
public Integer getPort() {
return port;
}
public void setPort(Integer port) {
this.port = port;
}
public String getUser() {
return user;
}
public void setUser(String user) {
this.user = user;
}
public String getPassword() {
return password;
}
public void setPassword(String password) {
this.password = password;
}
public Map<String, String> getFilePath() {
return filePath;
}
public void setFilePath(Map<String, String> filePath) {
this.filePath = filePath;
}
}
Service层：SftpService.java
package com.flower.integration.sftp;
import com.jcraft.jsch.ChannelSftp;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.file.support.FileExistsMode;
import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer;
import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.io.File;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
@Service("sftpService")
public class SftpService {
private Logger log = LoggerFactory.getLogger(this.getClass());
@Resource(name = "fileInChannel")
protected MessageChannel messageChannel;
@Autowired
private SftpProperty sftpProperty;
@Autowired
private SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory;
/**
* 发送文件到SFTP, 借用MessageChannel
*
* @param localFilePath file local path.
*/
public void sendFileToSftp(String localFilePath) {
Path filePath = Paths.get(localFilePath);
if (filePath.toFile().exists()) {
Message<File> fileMessage = MessageBuilder.withPayload(filePath.toFile()).build();
boolean result = messageChannel.send(fileMessage);
String resultMsg = result ? "Success" : "Failure";
log.info("File send to sftp {}, File: {}.", resultMsg, filePath.getFileName());
} else {
log.warn("No found file. {}", filePath.getFileName());
}
}
/**
* 删除sftp文件
*
* @param sessionFactory sftp server.
* @param remoteDirectory file directory.
* @param fileName file
* @return return true is remove success,or false.
*/
public boolean removeSftpRemoteFile(SessionFactory<ChannelSftp.LsEntry> sessionFactory, String remoteDirectory, String fileName) {
SftpRemoteFileTemplate sftpRemoteFileTemplate = new SftpRemoteFileTemplate(sessionFactory);
boolean direCheck = remoteDirectory.endsWith(sftpRemoteFileTemplate.getRemoteFileSeparator());
if (!direCheck) {
remoteDirectory += sftpRemoteFileTemplate.getRemoteFileSeparator();
}
boolean fileExist = sftpRemoteFileTemplate.exists(remoteDirectory + fileName);
if (fileExist) {
return sftpRemoteFileTemplate.remove(remoteDirectory + fileName);
} else {
log.warn("No found file in the directory, {}.", remoteDirectory);
return false;
}
}
/**
* sftp文件重命名
*
* @param sessionFactory sftp server
* @param remoteDirectory file directory path.
* @param sourceFileName source file name
* @param targetFileName rename target name
*/
public void renameSftpRemoteFile(SessionFactory<ChannelSftp.LsEntry> sessionFactory, String remoteDirectory,
String sourceFileName, String targetFileName) {
SftpRemoteFileTemplate fileTemplate = new SftpRemoteFileTemplate(sessionFactory);
boolean direCheck = remoteDirectory.endsWith(fileTemplate.getRemoteFileSeparator());
if (!direCheck) {
remoteDirectory += fileTemplate.getRemoteFileSeparator();
}
boolean fileExist = fileTemplate.exists(remoteDirectory + sourceFileName);
if (fileExist) {
fileTemplate.rename(remoteDirectory + sourceFileName, remoteDirectory + targetFileName);
} else {
log.warn("No found file in the directory, {}.", remoteDirectory);
}
}
/**
* sftp文件是否存在
*
* @param sessionFactory sftp server
* @param directory file directory
* @param fileName file name
* @return true if file exist, or false.
*/
public boolean fileExist(SessionFactory<ChannelSftp.LsEntry> sessionFactory, String directory, String fileName) {
SftpRemoteFileTemplate fileTemplate = new SftpRemoteFileTemplate(sessionFactory);
boolean fileNameCheck = directory.endsWith(fileTemplate.getRemoteFileSeparator());
if (!fileNameCheck) {
directory += fileTemplate.getRemoteFileSeparator();
}
return fileTemplate.exists(directory + fileName);
}
/**
* sftp检索文件
*
* @param sessionFactory sftp server
* @param directory file directory
* @param fileNameFilter file name filter
* @return file name list match filter
*/
public List<String> lsFileOfDirectory(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
String directory, String fileNameFilter) {
SftpRemoteFileTemplate fileTemplate = new SftpRemoteFileTemplate(sessionFactory);
if (!directory.endsWith(fileTemplate.getRemoteFileSeparator())) {
directory += fileTemplate.getRemoteFileSeparator();
}
ChannelSftp.LsEntry[] files = fileTemplate.list(directory + fileNameFilter);
List<String> fileNames = new ArrayList<>();
for (ChannelSftp.LsEntry lsEntry : files) {
boolean isDir = lsEntry.getAttrs().isDir();
if (!isDir) {
fileNames.add(lsEntry.getFilename());
}
}
return fileNames;
}
@Autowired
private BeanFactory beanFactory;
/**
* 本地发送文件至sftp服务器
*
* @param sessionFactory sftp server
* @param filePath file local path
* @param targetPath target directory
* @param mode FileExistsModel
* NULL：默认，替换文件；
* APPEND：若文件存在，追加内容；
* REPLACE：替换文件；
* APPEND_NO_FLUSH：
* FAIL：
* IGNORE：
*/
public void sendSftpFile(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
String filePath, String targetPath, FileExistsMode mode){
SftpRemoteFileTemplate fileTemplate = new SftpRemoteFileTemplate(sessionFactory);
try {
//设置远程sftp服务器配置
fileTemplate.setRemoteDirectoryExpression(new LiteralExpression(targetPath));
fileTemplate.setAutoCreateDirectory(true);
fileTemplate.setCharset("UTF-8");
fileTemplate.setBeanFactory(beanFactory);
fileTemplate.afterPropertiesSet();
} catch (Exception e){
log.warn(e.getMessage());
}
Path file = Paths.get(filePath);
if (file.toFile().exists()){
Message<File> message = MessageBuilder.withPayload(file.toFile()).build();
if (null == mode){
fileTemplate.send(message);
} else {
//fileTemplate.setFileNameGenerator(new DefaultFileNameGenerator());
if (fileTemplate.isUseTemporaryFileName()){
fileTemplate.setUseTemporaryFileName(false);
}
fileTemplate.send(message, mode);
}
}
}
@Resource(name = "synFileChannel")
private SftpInboundFileSynchronizer sftpInboundFileSynchronizer;
public void synchronizedFileToLocal(String localDir){
File dir = Paths.get(localDir).toFile();
sftpInboundFileSynchronizer.synchronizeToLocalDirectory(dir);
}
}
Controller层：用于测试service层方法
package com.flower.integration.sftp;
import com.jcraft.jsch.ChannelSftp;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.file.remote.FileInfo;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.io.File;
import java.util.List;
@RestController
public class TestController {
@Autowired
private SftpService sftpService;
@Autowired
private SftpConfig.UploadGateway uploadGateway;
@GetMapping("/sftp")
public void testSftpSpringBatch() {
List<FileInfo> fileList = uploadGateway.listFileInfo("/send");
for (FileInfo file : fileList) {
String fileName = file.getFilename();
String filePath = file.getRemoteDirectory();
ChannelSftp.LsEntry fileInfo = (ChannelSftp.LsEntry) file.getFileInfo();
boolean isDir = file.isDirectory();
boolean isLink = file.isLink();
long modifyTime = file.getModified();
System.out.println("============================= " + fileName);
System.out.println("================== " + filePath);
System.out.println("================== " + fileInfo.getFilename());
System.out.println("================== " + isDir);
System.out.println("================== " + isLink);
System.out.println("================== " + modifyTime);
}
}
@GetMapping("/sftp2")
public void testSftpSpringBatch2() {
List<FileInfo> fileNameList = uploadGateway.listFileName("/send");
for (FileInfo fileName : fileNameList) {
System.out.println("============================= " + fileName);
}
}
@GetMapping("/sftp3")
public void testSftpSpringBatch3() throws InterruptedException {
List<File> fileNameList = uploadGateway.listFile("/send");
for (File fileName : fileNameList) {
System.out.println("============================= " + fileName);
}
}
@GetMapping("/sftp4")
public void testSftpSpringBatch4() throws InterruptedException {
String result = uploadGateway.putFile(new File("G:\Redis.pdf"));
System.out.println("============================= " + result);
}
@GetMapping("/sftp5")
public void testSftpSpringBatch5() throws InterruptedException {
List<String> result = uploadGateway.mputFile(new File("G:\js"));
for (String fileName : result) {
System.out.println("============================= " + fileName);
}
}
@GetMapping("/sftp6")
public void testSftpSpringBatch6() throws InterruptedException {
boolean result = uploadGateway.removeFile("/send/2.txt");
System.out.println("============================= " + result);
}
@GetMapping("/sftp7")
public void testSftpSpringBatch7() throws InterruptedException {
boolean result = uploadGateway.moveFile("/22.TXT");
System.out.println("============================= " + result);
}
}
SpringIntegrationApp.java启动类
package com.flower.integration;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration;
import org.springframework.scheduling.annotation.EnableScheduling;
@SpringBootApplication
//@EnableScheduling
public class SpringIntegrationApp {
public static void main(String[] args) {
SpringApplication.run(SpringIntegrationApp.class, args);
System.out.println("Spring-Integration application start success.");
}
}
junit单元测试类：自行在test文件夹下建立并测试SpringIntegrationExamplesApplicationTests.java
package com.flower.integration.sftp;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringIntegrationExamplesApplicationTests {
@Test
public void contextLoads() {
}
}
进行单元测试SftpServiceTest.java
package com.flower.integration.sftp;
import com.flower.integration.SpringIntegrationApp;
import com.jcraft.jsch.ChannelSftp;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.file.support.FileExistsMode;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import java.util.List;
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = SpringIntegrationApp.class)
@EnableIntegration
public class SftpServiceTest {
@Autowired
private SftpService sftpService;
@Autowired
private SftpProperty sftpProperty;
@Autowired
private SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory;
@Before
public void before (){
System.out.println("00000000000000000000000000000000000000000000000000000");
}
@After
public void after(){
}
@Test
public void sendFileToSftp() {
//sftpService.sendFileToSftp();
}
@Test
public void testRemoveSftpRemoteFile(){
boolean result = sftpService.removeSftpRemoteFile(
sftpSessionFactory, sftpProperty.getSftpSendPath(),"user333.csv");
System.out.println("=======" + result);
}
@Test
public void testRenameSftpRemoteFile(){
sftpService.renameSftpRemoteFile(sftpSessionFactory, sftpProperty.getSftpSendPath(),"user.csv",
"user111.csv");
}
@Test
public void testfileExist(){
boolean result = sftpService.fileExist(sftpSessionFactory, sftpProperty.getSftpSendPath(),"user111.csv");
System.out.println("++++++++++++" + result);
}
@Test
public void testlsFileOfDirectory(){
List<String> result = sftpService.lsFileOfDirectory(sftpSessionFactory,
sftpProperty.getSftpSendPath(),"*TXT");
System.out.println("-------------------" + result.toString());
}
@Test
public void testSendSftpFile() throws Exception {
sftpService.sendSftpFile(sftpSessionFactory,
"G:\jquery.txt", sftpProperty.getSftpAchievePath(), FileExistsMode.REPLACE);
}
@Test
public void testSynchronizedFileToLocal(){
sftpService.synchronizedFileToLocal(sftpProperty.getLocalTempDir());
}
}