jsch的sftp在多线程下的问题及处理办法

作者	时间
雨中星辰	2022-02-09

jsch的sftp(ChannelSftp、Session)是不能在多线程下进行公用的，如果希望在多线程下操作sftp,那么ChannelSftp、Session需要放在ThreadLocal中。

单线程示例

SftpUtil2:

import com.jcraft.jsch.*;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Field;


/**
 * @author star
 * @descripton sftp工具类（非线程安全的）
 * @date 2021/6/10
 **/
@Slf4j
public class SftpUtil2 {
    Session session;
    ChannelSftp channel;

    public String username;
    public String password;
    public String remoteHost;
    public Integer remotePort;
    public String charset;

    public SftpUtil2(String remoteHost, Integer remotePort, String username, String password, String charset) throws JSchException, SftpException {
        this.remoteHost = remoteHost;
        this.remotePort = remotePort;
        this.username = username;
        this.password = password;
        this.charset = charset;
        connect();
    }

    public void uploadFile(ChannelSftp channel, String src, String dest) throws SftpException {
        channel.put(src, dest);
    }

    public void uploadFile(String src, String dest) throws SftpException, JSchException {
        uploadFile(channel, src, dest);
    }

    public void connect() throws JSchException, SftpException {
        JSch jSch = new JSch();
        session = jSch.getSession(username, remoteHost, remotePort);
        session.setPassword(password);
        session.setConfig("PreferredAuthentications", "password");
        session.setConfig("StrictHostKeyChecking", "no");// 为session重新设置参数
        session.connect();
        channel = (ChannelSftp) session.openChannel("sftp");
        channel.connect(5000);
        if (!"UTF-8".equalsIgnoreCase(charset)) {
            Class cl = ChannelSftp.class;
            Field f;
            try {
                f = cl.getDeclaredField("server_version");
                f.setAccessible(true);
                f.set(channel, 2);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        channel.setFilenameEncoding(charset);
    }

    public void disconnectAll() {
        if (channel != null && channel.isConnected()) {
            channel.disconnect();
        }
        if (session != null && session.isConnected()) {
            session.disconnect();
        }
    }

}

测试程序：

import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.SftpException;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.filefilter.TrueFileFilter;

import java.io.File;
import java.util.Collection;

/**
 * @author star
 * @date 2022/2/9 4:22 PM
 */
//1458个文件 每个500kb 耗时68807
//1458个文件 每个500kb 耗时68676
//1458个文件 每个500kb 耗时68714
public class Test1 {
    public static void main(String[] args) throws JSchException, SftpException {
        Collection<File> files = FileUtils.listFiles(new File("/tmp/test"), TrueFileFilter.INSTANCE, TrueFileFilter.INSTANCE);
        long start = System.currentTimeMillis();
        SftpUtil2 sftpUtil = new SftpUtil2("192.168.40.37", 22, "root", "R0ck9","UTF-8");
        for (File file : files) {
            sftpUtil.uploadFile(file.getPath(),"/tmp/aa");
        }
        System.out.println("上传完毕,耗时：" + (System.currentTimeMillis() - start));
        sftpUtil.disconnectAll();
    }
}

多线程错误示例

import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.SftpException;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.filefilter.TrueFileFilter;

import java.io.File;
import java.util.Collection;

/**
 * @author star
 * @date 2022/2/9 4:22 PM
 */
public class Test2 {
    public static void main(String[] args) throws JSchException, SftpException {
        Collection<File> files = FileUtils.listFiles(new File("/tmp/test"), TrueFileFilter.INSTANCE, TrueFileFilter.INSTANCE);
        SftpUtil2 sftpUtil = new SftpUtil2("192.168.40.37", 22, "root", "R0ck9","UTF-8");
        files.parallelStream().forEach(file -> {
            try {
                sftpUtil.uploadFile(file.getPath(),"/tmp/aa");
            } catch (SftpException | JSchException e) {
                e.printStackTrace();
            }
        });

        sftpUtil.disconnectAll();
    }
}

该程序使用多线程并发操作sftp,在多线程中共用一个ChannelSftp,就会出异常，具体如下：

Caused by: java.io.IOException: Pipe closed
    at java.io.PipedInputStream.read(PipedInputStream.java:307)
    at java.io.PipedInputStream.read(PipedInputStream.java:377)
    at com.jcraft.jsch.ChannelSftp.fill(ChannelSftp.java:2909)
    at com.jcraft.jsch.ChannelSftp.header(ChannelSftp.java:2935)
    at com.jcraft.jsch.ChannelSftp._put(ChannelSftp.java:583)
    ... 15 more
4: java.io.IOException: Pipe closed
    at com.jcraft.jsch.ChannelSftp._put(ChannelSftp.java:697)
    at com.jcraft.jsch.ChannelSftp.put(ChannelSftp.java:475)
    at com.jcraft.jsch.ChannelSftp.put(ChannelSftp.java:365)
    at SftpUtil2.uploadFile(SftpUtil2.java:33)
    at SftpUtil2.uploadFile(SftpUtil2.java:37)
    at Test2.lambda$main$0(Test2.java:19)
    at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
    at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
    at java.util.stream.ForEachOps$ForEachTask.compute(ForEachOps.java:291)
    at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.io.IOException: Pipe closed
    at java.io.PipedInputStream.read(PipedInputStream.java:307)
    at java.io.PipedInputStream.read(PipedInputStream.java:377)
    at com.jcraft.jsch.ChannelSftp.fill(ChannelSftp.java:2909)
    at com.jcraft.jsch.ChannelSftp.header(ChannelSftp.java:2935)
    at com.jcraft.jsch.ChannelSftp._put(ChannelSftp.java:583)
    ... 14 more
4: java.io.IOException: Pipe closed
    at com.jcraft.jsch.ChannelSftp._put(ChannelSftp.java:697)
    at com.jcraft.jsch.ChannelSftp.put(ChannelSftp.java:475)
    at com.jcraft.jsch.ChannelSftp.put(ChannelSftp.java:365)
    at SftpUtil2.uploadFile(SftpUtil2.java:33)
    at SftpUtil2.uploadFile(SftpUtil2.java:37)
    at Test2.lambda$main$0(Test2.java:19)
    at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
    at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
    at java.util.stream.ForEachOps$ForEachTask.compute(ForEachOps.java:291)
    at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.io.IOException: Pipe closed
    at java.io.PipedInputStream.read(PipedInputStream.java:307)
    at java.io.PipedInputStream.read(PipedInputStream.java:377)
    at com.jcraft.jsch.ChannelSftp.fill(ChannelSftp.java:2909)
    at com.jcraft.jsch.ChannelSftp.header(ChannelSftp.java:2935)
    at com.jcraft.jsch.ChannelSftp._put(ChannelSftp.java:583)
    ... 14 more

多线程正确示例

在SftpUtil中使用ThreadLocal包装ChannelSftp、Session保证线程安全。

在Test3中使用parallelStream().forEach完成多线程并发操作sftp。

SftpUtil

import com.jcraft.jsch.*;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Field;


/**
 * @author star
 * @descripton sftp工具类，线程安全的
 * @date 2021/6/10
 **/
@Slf4j
public class SftpUtil {

    ThreadLocal<ChannelSftp> channelThreadLocal = new ThreadLocal<>();
    ThreadLocal<Session> sessionThreadLocal = new ThreadLocal<>();

    public String username;
    public String password;
    public String remoteHost;
    public Integer remotePort;
    public String charset;

    public SftpUtil(String remoteHost, Integer remotePort, String username, String password, String charset) throws JSchException, SftpException {
        this.remoteHost = remoteHost;
        this.remotePort = remotePort;
        this.username = username;
        this.password = password;
        this.charset = charset;
        connect();
    }

    public void uploadFile(ChannelSftp channel, String src, String dest) throws SftpException {
        channel.put(src, dest);
    }

    public void uploadFile(String src, String dest) throws SftpException, JSchException {
        uploadFile(getChannel(), src, dest);
    }

    public void connect() throws JSchException, SftpException {
        JSch jSch = new JSch();
        Session session = jSch.getSession(username, remoteHost, remotePort);
        session.setPassword(password);
        session.setConfig("PreferredAuthentications","password");
        session.setConfig("StrictHostKeyChecking", "no");// 为session重新设置参数
        session.connect();
        ChannelSftp channel = (ChannelSftp) session.openChannel("sftp");
        channel.connect(5000);
        if (!"UTF-8".equalsIgnoreCase(charset)) {
            Class cl = ChannelSftp.class;
            Field f;
            try {
                f = cl.getDeclaredField("server_version");
                f.setAccessible(true);
                f.set(channel, 2);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        channel.setFilenameEncoding(charset);
        channelThreadLocal.set(channel);
        sessionThreadLocal.set(session);
    }

    public void disconnectAll() throws JSchException, SftpException {
        ChannelSftp channel = getChannel();
        if (channel != null && channel.isConnected()) {
            channel.disconnect();
        }
        Session session = getSession();
        if (session != null && session.isConnected()) {
            session.disconnect();
        }
    }

    public ChannelSftp getChannel() throws JSchException, SftpException {
        ChannelSftp channelSftp = channelThreadLocal.get();
        if (channelSftp == null) {
            connect();
        }
        return channelThreadLocal.get();
    }

    public Session getSession() throws JSchException, SftpException {
        Session session = sessionThreadLocal.get();
        if (session == null) {
            connect();
        }
        return sessionThreadLocal.get();
    }
}

Test3:

import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.SftpException;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.filefilter.TrueFileFilter;

import java.io.File;
import java.util.Collection;

/**
 * @author star
 * @date 2022/2/9 4:22 PM
 */
//1458个文件 每个500kb 耗时67506
//1458个文件 每个500kb 耗时66652
//1458个文件 每个500kb 耗时67193

public class Test3 {
    public static ThreadLocal<SftpUtil> sftpUtilThreadLocal = new ThreadLocal<>();

    public static void main(String[] args) throws JSchException, SftpException {
        Collection<File> files = FileUtils.listFiles(new File("/tmp/test"), TrueFileFilter.INSTANCE, TrueFileFilter.INSTANCE);
        long start = System.currentTimeMillis();
        SftpUtil sftpUtil = new SftpUtil("192.168.40.37", 22, "root", "R0ck9", "UTF-8");
        files.parallelStream().forEach(file -> {
            try {
                sftpUtil.uploadFile(file.getPath(), "/tmp/aa");
            } catch (SftpException | JSchException e) {
                e.printStackTrace();
            }
        });

        System.out.println("上传完毕,耗时：" + (System.currentTimeMillis() - start));
        sftpUtil.disconnectAll();
    }
}

性能测试

使用单线程上传1458个文件，每个500kb，进行三次测试，平均耗时为：68732毫秒

使用多线程上传1458个文件，每个500kb，进行三次测试，平均耗时为：67117毫秒

最佳实践

通过测试，可以看到使用多线程操作确实能提升效率，但是，其提升非常有限，却带来了较高的复杂性，在使用中更加建议通过单线程的方式。如果，需要将sftp封装成工具类供他人使用，需要提醒多线程并发的问题，或为了保险起见，也可以在sftp工具类中使用ThreadLocal包装ChannelSftp和Session。