java- SFTP文件上传下载

JSch - SFTP文件上传下载

文章目录

• JSch - SFTP文件上传下载
• • 1. JSch简介
  • 2. ChannelSftp常用Api
  • • JSch支持三种文件传输模式
    • 文件上传 put() 方法
    • 文件下载 get() 方法
  • 3. SFTP上传下载代码实现
  • • 1. sftp工具pom依赖
    • 2. 编写SFTP工具类
    • 3. 测试【文件上传】
    • 4. 测试【文件下载】
  • 4. SFTP监控传输进度
  • • 4.1 监控逻辑代码实现
    • 4.2 上传进度监控【测试】
    • 4.3 测试结果
  • 5. 扩展
  • • 1.配置文件
    • 2.文件信息实体
    • 3.SFTP工具类
    • 4.Linux命令工具类
    • 5.定义SFTP客户端
    • 6.文件传输测试
    • • 6.1 创建队列
      • 6.2 生产者
      • 6.3 消费者

1. JSch简介

​ JSch是Java Secure Channel的缩写，是一个java实现的可以完成sftp上传下载的工具，我们可以集成它的功能到自己的应用程序，本文介绍使用JSch实现的SFTP上传下载的功能。

2. ChannelSftp常用Api

ChannelSftp类是JSch实现SFTP核心类，它包含了所有SFTP的方法，如

JSch支持三种文件传输模式

参数中的文件传输模式可用 ChannelSftp类的静态成员变量

• ChannelSftp.OVERWRITE（完全覆盖模式）
• ChannelSftp.RESUME（恢复模式）
• ChannelSftp.APPEND（追加模式）

文件上传 put() 方法

put方法：将本地文件名为src的文件上传到目标服务器（12种重载）

注意：

• 将本地文件名为src的文件上传到目标服务器，目标文件名为dst，若dst为目录，则目标文件名将与src文件名相同。
• 可选参数：1.文件传输模式和 2.SftpProgressMonitor接口的monitor对象（查看上传进度条）

文件下载 get() 方法

get(): 将目标服务器上文件名为src的文件下载到本地

注意：des为本地文件名，若dst为目录，则下载到本地的文件名将与src文件名相同。

3. SFTP上传下载代码实现

​ 编写一个工具类，根据ip，用户名及密码得到一个SFTP channel对象，即ChannelSftp的实例对象，在应用程序中就可以使用该对象来调用SFTP的各种操作方法。

1. sftp工具pom依赖

     <dependency>          <groupId>com.jcraftgroupId>          <artifactId>jschartifactId>          <version>0.1.55version>     dependency>

2. 编写SFTP工具类

import java.util.Map;import java.util.Properties;import org.apache.log4j.Logger;import com.jcraft.jsch.Channel;import com.jcraft.jsch.ChannelSftp;import com.jcraft.jsch.JSch;import com.jcraft.jsch.JSchException;import com.jcraft.jsch.Session;public class SFTPChannel {    Session session = null;    Channel channel = null;    private static final Logger LOG = Logger.getLogger(SFTPChannel.class.getName());  //登录sftp    public ChannelSftp getChannel(Map<String, String> sftpDetails, int timeout) throws JSchException {        String ftpHost = sftpDetails.get(SFTPConstants.SFTP_REQ_HOST);        String port = sftpDetails.get(SFTPConstants.SFTP_REQ_PORT);        String ftpUserName = sftpDetails.get(SFTPConstants.SFTP_REQ_USERNAME);        String ftpPassword = sftpDetails.get(SFTPConstants.SFTP_REQ_PASSWORD);        int ftpPort = SFTPConstants.SFTP_DEFAULT_PORT;        if (port != null && !port.equals("")) {            ftpPort = Integer.valueOf(port);        }      //创建JSch对象        JSch jsch = new JSch();       //根据用户名，主机ip，端口获取一个Session对象        session = jsch.getSession(ftpUserName, ftpHost, ftpPort);         LOG.debug("Session created.");        if (ftpPassword != null) {            session.setPassword(ftpPassword);        }        Properties config = new Properties();        config.put("StrictHostKeyChecking", "no");        session.setConfig(config); // 为Session对象设置properties        session.setTimeout(timeout); // 设置timeout时间        session.connect(); // 通过Session建立链接        LOG.debug("Session connected.");        LOG.debug("Opening Channel.");        channel = session.openChannel("sftp"); // 打开SFTP通道        channel.connect(); // 建立SFTP通道的连接        LOG.debug("Connected successfully to ftpHost = " + ftpHost + ",as ftpUserName = " + ftpUserName                + ", returning: " + channel);        return (ChannelSftp) channel;    }  //退出sftp    public void closeChannel() throws Exception {        if (channel != null) {            channel.disconnect();        }        if (session != null) {            session.disconnect();        }    }}

SFTP静态成员变量类

public class SFTPConstants {    public static final String SFTP_REQ_HOST = "host";    public static final String SFTP_REQ_PORT = "port";    public static final String SFTP_REQ_USERNAME = "username";    public static final String SFTP_REQ_PASSWORD = "password";    public static final int SFTP_DEFAULT_PORT = 22;    public static final String SFTP_REQ_LOC = "location";}

3. 测试【文件上传】

import java.util.HashMap;import java.util.Map;import com.jcraft.jsch.ChannelSftp;public class SFTPTest {    public SFTPChannel getSFTPChannel() {        return new SFTPChannel();    }        public static void main(String[] args) throws Exception {        SFTPTest test = new SFTPTest();        Map<String, String> sftpDetails = new HashMap<String, String>();        // 设置主机ip，端口，用户名，密码        sftpDetails.put(SFTPConstants.SFTP_REQ_HOST, "10.9.167.55");        sftpDetails.put(SFTPConstants.SFTP_REQ_USERNAME, "root");        sftpDetails.put(SFTPConstants.SFTP_REQ_PASSWORD, "arthur");        sftpDetails.put(SFTPConstants.SFTP_REQ_PORT, "22");                String src = "D:\\DevSoft\\HB-SnagIt1001.rar"; // 本地文件名        String dst = "/home/omc/ylong/sftp/HB-SnagIt1001.rar"; // 目标文件名                      SFTPChannel channel = test.getSFTPChannel();        ChannelSftp chSftp = channel.getChannel(sftpDetails, 60000);                                chSftp.put(src, dst, ChannelSftp.OVERWRITE); // 代码段2（常用）                // chSftp.put(new FileInputStream(src), dst, ChannelSftp.OVERWRITE); // 代码段3                chSftp.quit();        channel.closeChannel();    }}

4. 测试【文件下载】

import java.io.FileOutputStream;import java.io.OutputStream;import java.util.HashMap;import java.util.Map;import com.jcraft.jsch.ChannelSftp;import com.jcraft.jsch.SftpATTRS;public class SFTPGetTest {    public SFTPChannel getSFTPChannel() {        return new SFTPChannel();    }    public static void main(String[] args) throws Exception {        SFTPGetTest test = new SFTPGetTest();        Map<String, String> sftpDetails = new HashMap<String, String>();        // 设置主机ip，端口，用户名，密码        sftpDetails.put(SFTPConstants.SFTP_REQ_HOST, "10.9.167.55");        sftpDetails.put(SFTPConstants.SFTP_REQ_USERNAME, "root");        sftpDetails.put(SFTPConstants.SFTP_REQ_PASSWORD, "arthur");        sftpDetails.put(SFTPConstants.SFTP_REQ_PORT, "22");                SFTPChannel channel = test.getSFTPChannel();        ChannelSftp chSftp = channel.getChannel(sftpDetails, 60000);                String filename = "/home/omc/ylong/sftp/INTPahcfg.tar.gz";        SftpATTRS attr = chSftp.stat(filename);        long fileSize = attr.getSize();                String dst = "D:\\INTPahcfg.tar.gz";        OutputStream out = new FileOutputStream(dst);        try {                        chSftp.get(filename, dst, new FileProgressMonitor(fileSize)); // 代码段1                        // chSftp.get(filename, out, new FileProgressMonitor(fileSize)); // 代码段2                                } catch (Exception e) {            e.printStackTrace();        } finally {            chSftp.quit();            channel.closeChannel();        }    }}

4. SFTP监控传输进度

JSch支持在文件传输时对传输进度的监控。可以实现JSch提供的SftpProgressMonitor接口来完成这个功能。

SftpProgressMonitor接口类的定义为：

package com.jcraft.jsch;public interface SftpProgressMonitor{  public static final int PUT=0;  public static final int GET=1;  void init(int op, String src, String dest, long max);  boolean count(long count);  void end();}

• init(): 当文件开始传输时，调用init方法。

• count(): 当每次传输了一个数据块后，调用count方法，count方法的参数为这一次传输的数据块大小。

• end(): 当传输结束时，调用end方法。

4.1 监控逻辑代码实现

每间隔一定的时间才获取一下文件传输的进度，看看下面的SftpProgressMonitor接口实现类：

import java.text.DecimalFormat;import java.util.Timer;import java.util.TimerTask;import com.jcraft.jsch.SftpProgressMonitor;//自定义SftpProgressMonitor接口的实现类public class FileProgressMonitor extends TimerTask implements SftpProgressMonitor {        private long progressInterval = 5 * 1000; // 默认间隔时间为5秒        private boolean isEnd = false; // 记录传输是否结束        private long transfered; // 记录已传输的数据总大小        private long fileSize; // 记录文件总大小        private Timer timer; // 定时器对象        private boolean isScheduled = false; // 记录是否已启动timer记时器      //构造方法中初始化文件大小，按需也可以添加其他参数    public FileProgressMonitor(long fileSize) {        this.fileSize = fileSize;    }        @Override    public void run() {        if (!isEnd()) { // 判断传输是否已结束            System.out.println("Transfering is in progress.");            long transfered = getTransfered();            if (transfered != fileSize) { // 判断当前已传输数据大小是否等于文件总大小                System.out.println("Current transfered: " + transfered + " bytes");                sendProgressMessage(transfered);            } else {                System.out.println("File transfering is done.");                setEnd(true); // 如果当前已传输数据大小等于文件总大小，说明已完成，设置end            }        } else {            System.out.println("Transfering done. Cancel timer.");            stop(); // 如果传输结束，停止timer记时器            return;        }    }        public void stop() {        System.out.println("Try to stop progress monitor.");        if (timer != null) {            timer.cancel();            timer.purge();            timer = null;            isScheduled = false;        }        System.out.println("Progress monitor stoped.");    }        public void start() {        System.out.println("Try to start progress monitor.");        if (timer == null) {            timer = new Timer();        }        timer.schedule(this, 1000, progressInterval);        isScheduled = true;        System.out.println("Progress monitor started.");    }            private void sendProgressMessage(long transfered) {        if (fileSize != 0) {            double d = ((double)transfered * 100)/(double)fileSize;            DecimalFormat df = new DecimalFormat( "#.##");             System.out.println("Sending progress message: " + df.format(d) + "%");        } else {            System.out.println("Sending progress message: " + transfered);        }    }        public boolean count(long count) {        if (isEnd()) return false;        if (!isScheduled) {            start();        }        add(count);        return true;    }        public void end() {        setEnd(true);        System.out.println("transfering end.");    }        private synchronized void add(long count) {        transfered = transfered + count;    }        private synchronized long getTransfered() {        return transfered;    }        public synchronized void setTransfered(long transfered) {        this.transfered = transfered;    }        private synchronized void setEnd(boolean isEnd) {        this.isEnd = isEnd;    }        private synchronized boolean isEnd() {        return isEnd;    }    public void init(int op, String src, String dest, long max) {        // Not used for putting InputStream    }}

4.2 上传进度监控【测试】

import java.io.File;import java.util.HashMap;import java.util.Map;import com.jcraft.jsch.ChannelSftp;public class SFTPTest {public SFTPChannel getSFTPChannel() {    return new SFTPChannel();}public static void main(String[] args) throws Exception {    SFTPTest test = new SFTPTest();    Map<String, String> sftpDetails = new HashMap<String, String>();    // 设置主机ip，端口，用户名，密码    sftpDetails.put(SFTPConstants.SFTP_REQ_HOST, "10.9.167.55");    sftpDetails.put(SFTPConstants.SFTP_REQ_USERNAME, "root");    sftpDetails.put(SFTPConstants.SFTP_REQ_PASSWORD, "arthur");    sftpDetails.put(SFTPConstants.SFTP_REQ_PORT, "22");        String src = "D:\\DevSoft\\HB-SnagIt1001.rar"; // 本地文件名    String dst = "/home/omc/ylong/sftp/HB-SnagIt1001.rar"; // 目标文件名              SFTPChannel channel = test.getSFTPChannel();    ChannelSftp chSftp = channel.getChannel(sftpDetails, 60000);        File file = new File(src);    long fileSize = file.length();                chSftp.put(src, dst, new FileProgressMonitor(fileSize), ChannelSftp.OVERWRITE); // 代码段2      // chSftp.put(new FileInputStream(src), dst, new FileProgressMonitor(fileSize), ChannelSftp.OVERWRITE); // 代码段3        chSftp.quit();    channel.closeChannel();}}

4.3 测试结果

logsTry to start progress monitor.Progress monitor started.Transfering is in progress.Current transfered: 98019 bytesSending progress message: 2.55%Transfering is in progress.Current transfered: 751479 bytesSending progress message: 19.53%Transfering is in progress.Current transfered: 1078209 bytesSending progress message: 28.02%......Transfering is in progress.Current transfered: 3430665 bytesSending progress message: 89.15%transfering end.Transfering done. Cancel timer.Try to stop progress monitor.Progress monitor stoped.

现在，程序每隔5秒钟才会打印一下进度信息。可以修改FileProgressMonitor类里的progressInterval变量的值，来修改默认的间隔时间。

5. 扩展

目标：zip压缩 + 加密 + SFTP

1.配置文件

#sftp server configsftp.url: 10.1.63.49sftp.port: 19222sftp.username: lihwsftp.password: 123456#是否压缩并加密：true/falseencryption: true#适配新版本ssh需添加对应的加密算法：true/falseisHightSSH: true#加密密钥zipPassword: xxx

2.文件信息实体

//保存要传输的文件基本信息public class TransferObject {    String localFilePath;    String localFileName;    String remoteFilePath;    String validTimeEnd;    private int retransTimes = 3;//失败重传次数    public TransferObject() {    }    public TransferObject(String localFilePath, String localFileName, String remoteFilePath, String validTimeEnd) {        this.localFilePath = localFilePath;        this.localFileName = localFileName;        this.remoteFilePath = remoteFilePath;        this.validTimeEnd = validTimeEnd;    }get/set/toString ....}

3.SFTP工具类

包含：文件上传、文件下载、构造JSch实体等方法（下面模板即插即用）

import com.jcraft.jsch.*;import org.apache.log4j.Logger;import java.io.*;import java.util.*;public class SftpClientUtil {    private static Logger logger = Logger.getLogger(SftpClientUtil.class);    private Session session;    private Channel channel;    private ChannelSftp sftp;    private InputStream in;    private OutputStream out;        public SftpClientUtil(String host, String username, String password, int port, boolean isHightSSH) throws Exception {        JSch jsch = new JSch();        this.session = jsch.getSession(username, host, port);        session.setPassword(password);        Properties config = new Properties();        config.put("StrictHostKeyChecking", "no"); // 不验证 HostKey        if (isHightSSH) {            config.put("kex", "diffie-hellman-group1-sha1,"                    + "diffie-hellman-group-exchange-sha1,"                    + "diffie-hellman-group-exchange-sha256"); //适配新版本ssh需添加对应的加密算法        }        session.setConfig(config);        try {            session.connect();        } catch (Exception e) {            if (session.isConnected())                session.disconnect();            logger.error("链接报错!!!", e);            throw new Exception("连接服务器失败,请检查主机[" + host + "],端口[" + port + "],用户名[" + username + "],端口[" + port + "]是否正确,以上信息正确的情况下请检查网络连接是否正常或者请求被防火墙拒绝.");        }        channel = session.openChannel("sftp");        try {            if (channel.isConnected())                channel.disconnect();            channel.connect();        } catch (Exception e) {            throw new Exception("连接服务器失败,请检查主机[" + host + "],端口[" + port + "],用户名[" + username + "],端口[" + port + "]是否正确,以上信息正确的情况下请检查网络连接是否正常或者请求被防火墙拒绝.");        }        sftp = (ChannelSftp) channel;    }        public SftpClientUtil(String host, String username, String password, int port, String encoding) throws Exception {        JSch jsch = new JSch();        this.session = jsch.getSession(username, host, port);        session.setPassword(password);        Properties config = new Properties();        config.put("StrictHostKeyChecking", "no"); // 不验证 HostKey        session.setConfig(config);        try {            session.connect();        } catch (Exception e) {            if (session.isConnected())                session.disconnect();            throw new Exception("连接服务器失败,请检查主机[" + host + "],端口[" + port + "],用户名[" + username + "],端口[" + port + "]是否正确,以上信息正确的情况下请检查网络连接是否正常或者请求被防火墙拒绝.");        }        channel = session.openChannel("sftp");        try {            channel.connect();        } catch (Exception e) {            if (channel.isConnected())                channel.disconnect();            throw new Exception("连接服务器失败,请检查主机[" + host + "],端口[" + port + "],用户名[" + username + "],密码[" + password + "]是否正确,以上信息正确的情况下请检查网络连接是否正常或者请求被防火墙拒绝.");        }        sftp = (ChannelSftp) channel;        sftp.setFilenameEncoding(encoding);    }        private SftpClientUtil(String host, String username, String password, int port, String encoding, int timeout)            throws Exception {        JSch jsch = new JSch();        this.session = jsch.getSession(username, host, port);        session.setPassword(password);        Properties config = new Properties();        config.put("StrictHostKeyChecking", "no"); // 不验证 HostKey        session.setConfig(config);        try {            session.connect();        } catch (Exception e) {            if (session.isConnected())                session.disconnect();            throw new Exception("连接服务器失败,请检查主机[" + host + "],端口[" + port + "],用户名[" + username + "],密码[" + password + "]是否正确,以上信息正确的情况下请检查网络连接是否正常或者请求被防火墙拒绝.");        }        session.setTimeout(timeout);        channel = session.openChannel("sftp");        try {            channel.connect();        } catch (Exception e) {            if (channel.isConnected())                channel.disconnect();            throw new Exception("连接服务器失败,请检查主机[" + host + "],端口[" + port + "],用户名[" + username + "],密码[" + password + "]是否正确,以上信息正确的情况下请检查网络连接是否正常或者请求被防火墙拒绝.");        }        sftp = (ChannelSftp) channel;        sftp.setFilenameEncoding(encoding);    }        public boolean uploadFile(String remotePath, String remoteFileName, String localPath, String localFileName) {        FileInputStream in = null;        try {            createDir(remotePath);            File file = new File(localPath + localFileName);            in = new FileInputStream(file);            sftp.put(in, remoteFileName);            return true;        } catch (FileNotFoundException e) {            e.printStackTrace();        } catch (SftpException e) {            e.printStackTrace();        } finally {            if (in != null) {                try {                    in.close();                } catch (IOException e) {                    e.printStackTrace();                }            }        }        return false;    }    //创建目录    public boolean createDir(String createpath) {        try {            createpath = createpath.trim();            if (isDirExist(createpath)) {                this.sftp.cd(createpath);                logger.info("cd " + createpath);                return true;            }            String[] pathArray = createpath.split("/");            for (String path : pathArray) {                path = path.trim();                if ("".equals(path)) {                    continue;                }                if (!isDirExist(path)) {                    sftp.mkdir(path);                }                logger.info("cd " + path);                sftp.cd(path);            }            return true;        } catch (SftpException e) {            e.printStackTrace();        }        return false;    }    //判断目录是否存在    public boolean isDirExist(String directory) {        boolean isDirExistFlag = false;        try {            SftpATTRS sftpATTRS = sftp.lstat(directory);            return sftpATTRS.isDir();        } catch (Exception e) {            if (e.getMessage().toLowerCase().equals("no such file")) {                logger.error("directory：" + directory + ",no such file ERROR!!！");            }        }        return isDirExistFlag;    }        public List<String> downloadFiles(String remotePath, String localPath){        List<String> downloadFiles = new ArrayList<String>();        try {            logger.info("切换到指定目录：" + remotePath);            boolean flag = openDir(remotePath, sftp);            if (flag) {                //1.获取远端路径下所有文件                Vector<?> vv = listFiles("*");                if (vv == null) {                    return null;                } else {                    for (Object object : vv) {                        ChannelSftp.LsEntry entry = (ChannelSftp.LsEntry) object;                        String remoteFileName = entry.getFilename();                        logger.info("校验文件名：" + remoteFileName);                        String path = localPath.substring(localPath.length() - 1).equals("/") ? localPath : localPath + "/";                        File file = new File(path + remoteFileName);                        logger.info("保存校对文件的本地路径为:" + file.getAbsolutePath());                        logger.info("start downLoad " + remoteFileName + " ~~");                        sftp.get(remoteFileName, new FileOutputStream(file));                        logger.info("downLoad ok ~~");                        downloadFiles.add(remoteFileName);                    }                    if (downloadFiles.size() < 1) {                        logger.error("remotePath：" + remotePath + "路径下，未匹配到校对文件！");                    }                }            } else {                logger.info("对应的目录" + remotePath + "不存在！");            }        } catch (Exception e) {            e.printStackTrace();        } finally {            if (sftp != null) {                if (sftp.isConnected()) {                    sftp.disconnect();                }            }            if (session != null) {                if (session.isConnected()) {                    session.disconnect();                }            }        }        return downloadFiles;    }    //打开或者进入指定目录    public boolean openDir(String directory, ChannelSftp sftp) {        try {            sftp.cd(directory);            logger.info("cd " + directory + " ok");            return true;        } catch (SftpException e) {            logger.error(e + "");            return false;        }    }    //返回目录下所有文件信息    public Vector listFiles(String directory) throws SftpException {        return sftp.ls(directory);    }        public void disconnect() {        try {            sftp.disconnect();        } catch (Exception ignored) {        }        try {            channel.disconnect();        } catch (Exception ignored) {        }        try {            session.disconnect();        } catch (Exception ignored) {        }    }        public static SftpClientUtil getInstans(String host, String username, String password, int port) throws Exception {        return new SftpClientUtil(host, username, password, port, false);    }        public static SftpClientUtil getInstans(String host, String username, String password, int port, String encoding)            throws Exception {        return new SftpClientUtil(host, username, password, port, encoding);    }        public static SftpClientUtil getInstans(String host, String username, String password, int port, String encoding,                int timeout) throws Exception {        return new SftpClientUtil(host, username, password, port, encoding, timeout);    }}

4.Linux命令工具类

import org.apache.log4j.Logger;import java.io.BufferedReader;import java.io.IOException;import java.io.InputStream;import java.io.InputStreamReader;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.ThreadPoolExecutor;public class ExecUtils {    public static Logger logger = Logger.getLogger(ExecUtils.class);    private static final ExecutorService THREAD_POOL = (ThreadPoolExecutor) Executors.newFixedThreadPool(2);    public static boolean exec(String cmd) throws IOException, InterruptedException {        String[] cmds = {"/bin/sh", "-c", cmd};        Process process = Runtime.getRuntime().exec(cmds);        //消费正常日志        StringBuffer result = clearStream(process.getInputStream());        //消费错误日志        StringBuffer errorInfo = clearStream(process.getErrorStream());        //i为返回值，判断是否执行成功        int i = process.waitFor();        if (i != 0) {            return false;        }        return true;    }    private static StringBuffer clearStream(final InputStream stream) {        final StringBuffer result = new StringBuffer();        //处理buffer的线程        THREAD_POOL.execute(new Runnable() {            @Override            public void run() {                String line;                BufferedReader in = null;                try {                    in = new BufferedReader(new InputStreamReader(stream));                    while ((line = in.readLine()) != null) {                        result.append(line).append("\n");                    }                } catch (IOException e) {                    logger.error("error exec shell.", e);                } finally {                    if (in != null) {                        try {in.close();                        } catch (IOException ignored) {                        }                    }                }            }        });        return result;    }}

5.定义SFTP客户端

import com.ebupt.crbt.model.TransferObject;import org.apache.log4j.Logger;import org.springframework.beans.factory.annotation.Value;import org.springframework.stereotype.Component;import java.io.IOException;@Componentpublic class SFTPClientWrapper {    public Logger logger = Logger.getLogger(SFTPClientWrapper.class);    @Value("${sftp.url}")    private String url;    @Value("${sftp.port}")    private Integer port;    @Value("${sftp.username}")    private String username;    @Value("${sftp.password}")    private String password;    @Value("${isHightSSH}")    private Boolean isHightSSH; //jsch 与 ssh 版本适配，ssh>7.6 为true,反之为false    @Value("${encryption}")    private Boolean encryption; //是否加密：yes/no    @Value("${zipPassword}")    private String provinceID; //加密秘钥    private SftpClientUtil sftpClientUtil_www;    get/set...      //文件压缩（zip）+ 文件上传【在业务层调用时，将文件信息封装到TransferObject实体中】    public void zipAndUpload(TransferObject transferObject) throws Exception {        String localFileName = transferObject.getLocalFileName();        if (encryption && !localFileName.contains(".zip")) {            // 需要压缩，但未压缩            String remoteFileName = localFileName.replace(".dat", ".zip");            if (zip(transferObject.getLocalFilePath(), localFileName, remoteFileName, transferObject.getValidTimeEnd())) {                transferObject.setLocalFileName(remoteFileName);            }        }        uploadFileSftp(url, port, username, password, transferObject.getLocalFilePath(), transferObject.getRemoteFilePath(), transferObject.getLocalFileName());    }    private boolean zip(String localFilePath, String localFileName, String remoteFileName, String validTimeEnd) throws IOException, InterruptedException {        //压缩密码，配置文件中可配        String password ="xxx";        String zipCmd = "zip -q -j -D -P " + password + " " + localFilePath + remoteFileName + " " + localFilePath + localFileName;        return ExecUtils.exec(zipCmd);    }      private void uploadFileSftp(String host, int port, String userName, String password, String localPath,    String remotePath, String localFileName) throws Exception {        try {            if (sftpClientUtil_www != null) {                sftpClientUtil_www.disconnect();// 避免不同目录切换问题            }            sftpClientUtil_www = new SftpClientUtil(host, userName, password, port, isHightSSH);//Boolean.parseBoolean(isHighSSH)            sftpClientUtil_www.uploadFile(remotePath, localFileName, localPath, localFileName);        } catch (Exception e) {            logger.info("sftp upload file error." + localFileName, e);            if (sftpClientUtil_www != null) {                sftpClientUtil_www.disconnect();            }            throw e;        }    }}

6.文件传输测试

6.1 创建队列

//要sftp上传的文件信息添加到消息队列@Servicepublic class FileUpdateQueue {public static final ConcurrentLinkedQueue<TransferObject> uploadList = new ConcurrentLinkedQueue<TransferObject>();    public void upload(String localFilePath, String localFileName, String remoteFilePath, String validTimeEnd)            throws Exception {        if (localFilePath == null || localFileName == null || remoteFilePath == null || validTimeEnd == null) {            return;        }        if (localFileName.contains("_END_")) {            return;        }        TransferObject transferObject = new TransferObject(localFilePath, localFileName, remoteFilePath, validTimeEnd);        logger.info("add file to upload list." + transferObject.toString());        uploadList.offer(transferObject);    }}

6.2 生产者

@Componentpublic class SyncJob {    @Autowired  private FileUpdateQueue fileQueue;    @Scheduled(cron = "${job.scheduled}")  public void sync() throws Exception {        //xxx巴拉巴拉巴拉 按需进行一堆处理     try {      //文件上传【添加到消息队列ConcurrentLinkedQueue】fileQueue.upload(本地文件路径, 文件名, 远端文件路径, 文件名);} catch (Exception e) {e.printStackTrace();}    }    }

6.3 消费者

@Componentpublic class uploadFileJob {    private static final Logger logger = LoggerFactory.getLogger(uploadFileJob.class);    @Autowired    private SFTPClientWrapper sftp;    @Scheduled(cron = "${job.uploadFileScheduled}")    public void uploadFileWork() {        logger.info("##################### start to upload file #####################");        int size = FileUpdateQueue.uploadList.size();        logger.info("##################### current file num = " + size + "#####################");        TransferObject transferObject;        for (int i = 0; i < size; i++) {            transferObject = FTPClientWrapper.uploadList.poll();            try {                if (transferObject != null && transferObject.getRetransTimes() > 0) {                    logger.info("start sftp upload." + transferObject.toString());                    sftp.zipAndUpload(transferObject);                }            } catch (Exception e) {                logger.info("sftp upload error." + transferObject.toString(), e);                FTPClientWrapper.uploadList.offer(transferObject);            } finally {                if (transferObject != null) {                    transferObject.decrementRetransTimes();                }            }        }        logger.info("##################### start to upload file. current file num = " + FTPClientWrapper.uploadList.size() + " #####################");    }}

文章出处：http://www.cnblogs.com/longyg/archive/2012/06/25/2556576.html

来源地址：https://blog.csdn.net/weixin_44783506/article/details/129271770