java - 使用JSch使用多线程下载文件

Question

我正在尝试使用多线程从远程服务器下载多个文件。但是，当我使用多个线程时，我得到了

java.lang.IOException : Pipes closed.

当我只使用一个线程时，相同的代码工作正常。

难道不能使用 JSch 从同一个远程服务器同时下载多个文件吗？

SftpTest.java

public class SftpTest {

private static List<SftpAccessor> accessorList = new ArrayList<SftpAccessor>();
private static List<Payload> files = new ArrayList<Payload>();
private static ExecutorService writerThreadPool = Executors.newFixedThreadPool(10);

public static void main(String args[]) throws JSchException, InterruptedException {
    SSH srcServer = new SSH();
    srcServer.setHostname("10.22.65.140");
    srcServer.setKey("D:\\jars\\dmwvmcol01.ec2user.pem");
    srcServer.setPort("22");
    srcServer.setUsername("ec2-user");

    SftpAccessor acc = new SftpAccessor(srcServer);
    accessorList.add(acc);

    files.addAll(acc.ls("/data/test/src", false, "*", 1));

    for (Payload file : files) {
        writerThreadPool.submit(new LocalWriterThread(file));
    }

    writerThreadPool.shutdown();
    writerThreadPool.awaitTermination(20, TimeUnit.MINUTES);
}}

LocalWriterThread.java

public class LocalWriterThread implements Callable<String> {

private Payload payload;

public LocalWriterThread(Payload payload) {
    this.payload = payload;
}

public String call() throws Exception {
    String dest = "D:\\output\\jobsystemnew";
    System.out.println(payload.getFilename());
    File file = new File(dest + "/" + payload.getFilename());
    file.createNewFile();
    FileOutputStream fos = new FileOutputStream(file);
    copy(payload.getInputStream(), fos);
    payload.setInputStream(null);
    return "";
}

private void copy(InputStream is, OutputStream os) throws IOException {

    BufferedInputStream bis = null;
    BufferedOutputStream bos = null;
    try {
        System.out.println("Started Copying file : " + payload.getFilename());

        bis = new BufferedInputStream(is);
        bos = new BufferedOutputStream(os);

        byte[] buffer = new byte[1024];

        while (bis.read(buffer) != -1) {
            bos.write(buffer);
        }

        bos.flush();
        System.out.println("Finished Copying file : " + payload.getFilename());
    }
    finally {
        System.out.println("Closing input stream for " + payload.getFilename());
        bis.close();
        bos.close();
        /*
         * try { if (is != null) { is.close(); } } catch (Exception e) { System.out.println(" " + e + " : " + e.getMessage() + " unable to close input stream : " + payload.getFilename());
         * e.printStackTrace(); }
         */
        /*
         * try { if (os != null) { os.close(); } } catch (Exception e) { System.out.println(" " + e + " : " + e.getMessage() + " unable to close output stream : " + payload.getFilename());
         * e.printStackTrace(); }
         */
    }

}}

SSH.java

public class SSH {
String hostname;
String port;
String username;
String key;

public String getHostname() {
    return hostname;
}
public void setHostname(String hostname) {
    this.hostname = hostname;
}
public String getPort() {
    return port;
}
public void setPort(String port) {
    this.port = port;
}
public String getUsername() {
    return username;
}
public void setUsername(String username) {
    this.username = username;
}
public String getKey() {
    return key;
}
public void setKey(String key) {
    this.key = key;
} }

SftpAccessor.java

public class SftpAccessor {

private JSch jsch = new JSch();
private ChannelSftp channelSftp = null;
private Session session = null;

String errorMsg = null;

public SftpAccessor(SSH ssh) throws JSchException {

    jsch.addIdentity(ssh.getKey());
    session = jsch.getSession(ssh.getUsername(), ssh.getHostname(), Integer.parseInt(ssh.getPort()));
    session.setHostKeyAlias(ssh.getKey());

    java.util.Properties config = new java.util.Properties();
    config.put("StrictHostKeyChecking", "no");
    session.setConfig(config);
    session.connect();
    channelSftp = (ChannelSftp) session.openChannel("sftp");
    channelSftp.connect();

}

public List<Payload> ls(String directory, boolean recursive, String filemask, int delayMinutes) {

    List<Payload> files = new ArrayList<Payload>();

    try {

        channelSftp.cd(directory);

        @SuppressWarnings("unchecked")
        Vector<ChannelSftp.LsEntry> entries = channelSftp.ls(filemask);

        for (ChannelSftp.LsEntry entry : entries) {
            try {
                if (entry.getAttrs().isDir()) {
                    if (recursive) {
                        List<Payload> filesToAdd = ls(directory + "/" + entry.getFilename(), recursive, filemask, delayMinutes);
                        files.addAll(filesToAdd);
                    }
                }
                else {

                    Date lastmodified = new Date(entry.getAttrs().getMTime() * 1000L);
                    Date currdate = new Date(new Date().getTime() - (delayMinutes * 60 * 1000L));

                    if (lastmodified.before(currdate)) {
                        String filename = entry.getFilename();
                        entry.getAttrs().getMTime();

                        Payload file = new Payload();
                        System.out.println("Getting input Stream for " + directory + "/" + filename);
                        file.setInputStream(channelSftp.get(directory + "/" + filename));
                        file.setFilename(filename);
                        file.setLastModified(lastmodified);
                        files.add(file);
                    }

                }
            }
            catch (Exception e) {
                e.printStackTrace();
            }
        }

    }
    catch (SftpException e) {

    }       

    return files;
}}

有效载荷.java

public class Payload {

private String filename;
private InputStream inputStream;
private Date lastModified;

public String getFilename() {
    return filename;
}

public void setFilename(String filename) {
    this.filename = filename;
}

public Date getLastModified() {
    return lastModified;
}

public void setLastModified(Date lastModified) {
    this.lastModified = lastModified;
}

public InputStream getInputStream() {
    return inputStream;
}

public void setInputStream(InputStream inputStream) {
    this.inputStream = inputStream;
}}

Answer 1

JSch 不是线程安全的。即使是这样，通过单个 SSH session 使用并行下载也几乎没有任何性能优势。它会像串行下载一样慢。此外，您可能会达到并发文件句柄的服务器端限制。

您应该为每个线程/下载打开一个单独的 session 。