使用ftp，sftp多线程给各省同步文件信息一

74 阅读 0 评论 49 点赞

我是靠谱客的博主安静小蝴蝶，最近开发中收集的这篇文章主要介绍使用ftp，sftp多线程给各省同步文件信息一，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

开发环境如果没有ftp 服务器的话，可以下载一个serv-U 挺好用的；运行起来就行一个本地的ftp服务器；

1.需求 给各省根据配置表以ftp或sftp协议的方式，多线程定时任务给各省同步所需的数据；

设计表：业务基本配置表：配置对外提供哪些业务；

省份业务关系表：配合业务和省份的关系，配置哪些省份需要同步哪些业务数据；

省份业务模板关系表：针对模板信息，根据省份业务以及模板配置对外提供的哪些数据；

省份文件路径表：配置省份及 ftp或sftp 的文件夹路径及传输协议；

省份 ip 端口号用户名密码表：记录ftp sftp的ip端口用户名密码等信息；用来初始化链接；

等等等表....

2 ,设计：ftp sftp 的链接创建设计6个类：涉及接口，抽象类，工程类，

优点：需要增加传输协议例如本来只是ftp后来增加sftp 或者htttp等，只需要增加实现类就可以；不需要改动之前的代码逻辑；

1顶级抽象类：
/**
* Created by 12 on 2018/8/11.
*/
@Component
public abstract class AbstractFTPFileTransfer implements FileTransferInterface{

}

2接口类：定义通用方法：

public interface FileTransferInterface {
public void init(HashMap connectInfo);
public void upload(String fileName, InputStream fileInputStream);
public void rename(String ftppathName,String newftpPathName);
public void close();

}

3工厂类：提供初始化链接以及提供一些接口；

@Component
public class FTPTransterFactory {
private static final Logger logger = LoggerFactory.getLogger(FTPTransterFactory.class);
@Reference
private IKmSyncFtpPathSV iKmSyncFtpPathSV;
@Reference
private IKmSyncCollectLogSV iKmSyncCollectLogSV;

@Reference
private IKmSyncUploadLogSV iKmSyncUploadLogSV;

@Reference
private IKmTaskSV kmTaskSV;

// public Map<String, FileTransferInterface> getFileTransferInterface(List<HashMap<String, String>> map) {
FileTransferInterface ftpInfotr = AppContext.getBean((String) ftpInfo.getTrsfAgrmt());
ftpInfotr.init();
// return new HashMap<String, FileTransferInterface>();
// }
HashMap connectInfo;
public FileTransferInterface getFileTransferInterface(HashMap pathAndProvince) {
String pathId = String.valueOf(pathAndProvince.get("pathId"));
connectInfo = iKmSyncFtpPathSV.getConnectInfo(pathId);
String connectClassName = String.valueOf(connectInfo.get("connectClassName"));
FileTransferInterface connectObj = AppContext.getBean(connectClassName);
return connectObj;
}

public HashMap getConnectInfo(){
return connectInfo;
}

public IKmSyncCollectLogSV getIkmSyncCollectLogSv(){
return iKmSyncCollectLogSV;
}

public IKmSyncUploadLogSV getiKmSyncUploadLogSV(){
return iKmSyncUploadLogSV;
}

public IKmTaskSV getKmTaskSV(){return kmTaskSV;}
}

4，5 ftp 和sftp 的具体类：因为涉及多线程调用，所以需要加@scope

@Component("fTPFileTransfer")
@Scope(value = "prototype")
public class FTPFileTransfer extends AbstractFTPFileTransfer {
private static final com.cmos.core.logger.Logger logger = LoggerFactory.getLogger(FTPFileTransfer.class);

private FtpUtil ftpUtil = null;
private FTPClient ftpClient;
@Override
public void init(HashMap connectInfo){
try{
String ipValue = String.valueOf(connectInfo.get("ipValue"));
int portValue = Integer.parseInt(String.valueOf(connectInfo.get("portValue")));
String userName = String.valueOf(connectInfo.get("userName"));
String password = String.valueOf(connectInfo.get("password"));
String catalogPath = String.valueOf(connectInfo.get("catalogPath"));
ftpUtil = new FtpUtil();
ftpClient = ftpUtil.initFtpClient(ipValue,portValue,userName,password);
boolean changeFlag = ftpClient.changeWorkingDirectory(catalogPath);
if (!changeFlag){
logger.error("ftp链接初始化，变更到指定目录失败");
return;
}
}catch (Exception e){
logger.error("ftp链接初始化："+e);
}
}

@Override
public void upload(String tempFileName, InputStream fileInputStream) {
try {
ftpClient.storeFile(tempFileName,fileInputStream);
}catch (Exception e){
logger.error("ftp上传文件异常："+e);
}
}

@Override
public void rename(String tempFileName, String fileName) {
try {
ftpClient.rename(tempFileName, fileName);
}catch (Exception e){
logger.error("ftp修改文件名称异常"+e);
}
}

@Override
public void close() {
try {
ftpClient.logout();
ftpClient.disconnect();
}catch (Exception e ){
logger.error("ftp退出异常"+e);
}
}
}

@Component("sFTPFileTransfer")
@Scope(value = "prototype")
public class SFTPFileTransfer extends AbstractFTPFileTransfer{

private static final com.cmos.core.logger.Logger logger = LoggerFactory.getLogger(SFTPFileTransfer.class);

private MySFtp mySFtp = null;
private ChannelSftp channelSftp = null;

@Override
public void init(HashMap connectInfo) {
try{
String ipValue = String.valueOf(connectInfo.get("ipValue"));
int portValue = Integer.parseInt(String.valueOf(connectInfo.get("portValue")));
String userName = String.valueOf(connectInfo.get("userName"));
String password = String.valueOf(connectInfo.get("password"));
String catalogPath = String.valueOf(connectInfo.get("catalogPath"));
mySFtp = new MySFtp();
channelSftp = mySFtp.connect(ipValue,portValue,userName,password);
channelSftp.cd(catalogPath);
}catch (Exception e){
logger.error("sftp链接初始化："+e);
}
}

@Override
public void upload(String tempFileName, InputStream fileInputStream) {
try {
channelSftp.put(fileInputStream,tempFileName);
}catch (Exception e){
logger.error("sftp上传文件异常："+e);
}
}

@Override
public void rename(String tempFileName, String fileName) {
try {
channelSftp.rename(tempFileName,fileName);
}catch (Exception e){
logger.error("sftp修改文件名称异常"+e);
}
}

@Override
public void close() {
try {
channelSftp.exit();
}catch (Exception e ){
logger.error("sftp退出异常"+e);
}
}
}
6 ftp sftp 工具类：

public class FtpUtil {

private static final Logger logger = LoggerFactory.getLogger(FtpUtil.class);
private static FTPClient ftpClient = null;
private String _host = null;
private int _port = 0;
private String _username = null;
private String _password = null;
private String _remotePath = null;
public static final String FILE_SPEARATROR = "/"; // 路径分隔符
private static String LOCAL_CHARSET = "GBK"; // 本地字符编码
private static String SERVER_CHARSET = "ISO-8859-1";// FTP协议里面，规定文件名编码为iso-8859-1

public FTPClient initFtpClient(String ip, int port, String username, String password) {
long start = System.currentTimeMillis();
ftpClient = new FTPClient();
ftpClient.setControlEncoding("utf-8");
try {
ftpClient.connect(ip, port);
//登录ftp服务器
ftpClient.login(username, password);
//是否成功登录服务器
int replyCode = ftpClient.getReplyCode();
logger.info("reply == " + replyCode);
//设置被动模式
ftpClient.enterLocalPassiveMode();
ftpClient.setFileType(FTP.BINARY_FILE_TYPE);
if(!FTPReply.isPositiveCompletion(replyCode)){
ftpClient.disconnect();
logger.error("connect failed...ftp服务器:" + ip + ":" + port);
return null;
} else {
logger.info("connect successful...ftp服务器:" + ip + ":" + port);
}
long end = System.currentTimeMillis();
logger.info("ftp connect time:" +(end - start));

} catch (IOException e) {
logger.error("ftp连接失败：" + ip + ":" + port, e);
return null;
}
return ftpClient;
}

//判断ftp服务器文件是否存在
public boolean existFile(String path) {
boolean flag = false;
FTPFile[] ftpFileArr;
try {
ftpFileArr = ftpClient.listFiles(path);
} catch (IOException e) {
logger.error("读取文件失败");
return flag;
}
if (ftpFileArr.length > 0) {
flag = true;
}
return flag;
}

/**
* 退出ftp
*/
public void logout() {
try {
ftpClient.logout();
ftpClient.disconnect();
} catch (IOException e) {
logger.error("logout ftp fail ....." + e.getMessage(), e);
}
}

/**
* 空的构造方法
*/
public FtpUtil(){
}

/**
* 创建实例对象
* @param hostIp
* @param port
* @param username
* @param password
* @throws IOException
*/
public FtpUtil(String hostIp, int port, String username, String password) throws IOException {
logger.debug("hostIp --> " + hostIp + " , port -->" + port + " , username -->" + username + " , password -->" + password);
initFtpClient(hostIp,port,username,password);
_host = hostIp;
_port = port;
_username = username;
_password = password;
}

/**
* 创建一个带路径的实例对象
* @param hostIp
* @param port
* @param username
* @param password
* @param remotepath
* @throws IOException
*/
public FtpUtil(String hostIp, int port, String username, String password, String remotepath) throws IOException {
logger.debug("hostIp --> " + hostIp + " , port -->" + port + " , username -->" + username + " , password -->" + password + " , remotepath -->" + remotepath);
initFtpClient(hostIp,port,username,password);
ftpClient.changeWorkingDirectory(remotepath);
_host = hostIp;
_port = port;
_username = username;
_password = password;
_remotePath = remotepath;
}

/**
* 上传文件
* @param remotePath 路径
* @param filename 文件名
* @param input 文件流
* @return
*/
public boolean uploadFile(String remotePath, String filename,InputStream input) {
boolean result = false;
try {
if(!ftpClient.isConnected()){
this.forceReconnect();
}
// 检验是否连接成功
boolean isConnected = this.isConnected();
//注意：有的FTP是需要先发送CLNT版本号的
if (FTPReply.isPositiveCompletion(ftpClient.sendCommand("OPTS UTF8", "ON"))) {// 开启服务器对UTF-8的支持，如果服务器支持就用UTF-8编码，否则就使用本地编码（GBK）.
setLocalCharset("UTF-8");
}
ftpClient.setControlEncoding(LOCAL_CHARSET);
ftpClient.enterLocalPassiveMode();// 设置被动模式
ftpClient.setFileType(FTP.BINARY_FILE_TYPE);
// 判断链接是否可用
if(ftpClient.isAvailable()){
// 切换到正确的目录，无则创建
changeToRightDir(remotePath);

// 上传
result = ftpClient.storeFile(new String(filename.getBytes(LOCAL_CHARSET), SERVER_CHARSET), input);
if (result) {

}
}
input.close();
ftpClient.logout();
} catch (IOException e) {
logger.error(e.getMessage(), e);
} finally {
if (ftpClient.isConnected()) {
try {
ftpClient.disconnect();
} catch (IOException ioe) {
logger.error(ioe.getMessage(), ioe);
}
}
}
return result;
}

public void reconnect(){
if (!ftpClient.isConnected()) {
try {
FtpUtil.setConnected();
ftpClient.connect(_host, _port);
ftpClient.login(_username, _password);
ftpClient.changeWorkingDirectory(_remotePath);
} catch (IOException e) {
logger.error("",e);
}
}
}

public void forceReconnect() throws IOException {
if (!ftpClient.isConnected()) {
ftpClient.disconnect();
}
FtpUtil.setConnected();
ftpClient.connect(_host, _port);
ftpClient.login(_username, _password);
ftpClient.changeWorkingDirectory(_remotePath);
}

/**
* 强制更换目录到目标目录
* @param pathName
* @throws IOException
*/
public void changeWorkingDirectory(String pathName) throws IOException {
if (FtpUtil.ftpClient != null) {
try {
FtpUtil.ftpClient.changeWorkingDirectory(pathName);
} catch (IOException e) {
logger.error(e.getMessage(), e);
throw new IOException(e.getMessage());
}
}
}

/**
* 获取当前的工作目录
* @return
* @throws IOException
*/
public String getWorkingDirectory() throws IOException {
if (FtpUtil.ftpClient != null) {
return FtpUtil.ftpClient.printWorkingDirectory();
}
return null;
}

/**
* 切换到指定目录(不存在就创建目录同时切换到新建的目录)
* @param directoryId
* @throws Exception
*/
public void changeToRightDir(String directoryId) throws IOException {
if (ftpClient != null) {
String ftpDir;
try {
ftpDir = ftpClient.printWorkingDirectory();
} catch (IOException e) {
logger.error(e.getMessage(), e);
throw new IOException(e.getMessage());
}
if (ftpDir == null) {
ftpDir = "/";
}
String curDir;
if (directoryId.startsWith("/")) {
curDir = directoryId;
} else {
if (!ftpDir.endsWith("/")) {
curDir = ftpDir + FILE_SPEARATROR + directoryId;
} else {
curDir = ftpDir + directoryId;
}
}

boolean flag;
try {
flag = ftpClient.changeWorkingDirectory(curDir);
} catch (IOException e) {
logger.error(e.getMessage(), e);
throw new IOException(e.getMessage());
}
if (flag == false) {
try {
ftpClient.makeDirectory(directoryId);
ftpClient.changeWorkingDirectory(curDir);
} catch (IOException e) {
logger.error(e.getMessage(), e);
throw new IOException(e.getMessage());
}
}
}
}

/**
* 校验连接是否正常
* @return
* @throws IOException
*/
public boolean isConnected() throws IOException {
// 检验是否连接成功
int reply = ftpClient.getReplyCode();
if (!FTPReply.isPositiveCompletion(reply)) {

ftpClient.disconnect();
return false;
}
return true;
}

private static void setLocalCharset(String localCharset) {
LOCAL_CHARSET = localCharset;
}

private static void setConnected() throws IOException {
FTPClient tempClient = new FTPClient();
ftpClient = tempClient;
}
}

ok，准备就绪之后，剩下的就只有业务层面的代码逻辑了；时间有限，不忙了补充下一篇；