使用jsch和commons-pool2为sftp创建连接池

1、引入jsch和common-pool2包

以下为gradle的引入方式，使用maven的可切换为maven的xml引入方式。

implementation group: 'com.jcraft', name: 'jsch', version: "0.1.55"
implementation group: 'org.apache.commons', name: 'commons-pool2', version: "2.11.1"
1
2

2、创建SFTP操作对象，内部包含了各种操作方法

package cn.ac.iscas.dmo.db.sftp;

import cn.ac.iscas.dmo.db.api.util.StringUtils;
import cn.ac.iscas.dmo.db.filesystem.base.util.FsCommonUtils;
import cn.ac.iscas.dmo.db.sftp.pool.PoolProp;
import com.jcraft.jsch.*;

import java.io.*;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Iterator;
import java.util.Vector;

/**
 * @author zhuquanwen
 * @version 1.0
 * @date 2022/9/8 9:43
 * @since jdk11
 */
public class Sftp {
    private ChannelSftp channelSftp;
    private Session session;
    private String oriPath;

    public Sftp(PoolProp poolProp) {
        try {
            JSch jsch = new JSch();
            session = jsch.getSession(poolProp.getUsername(), poolProp.getHost(), poolProp.getPort());
            session.setPassword(poolProp.getPassword());
            session.setConfig("StrictHostKeyChecking", "no");
            session.connect(poolProp.getSessionConnectTimeout());
            Channel channel = session.openChannel("sftp");
            channel.connect(poolProp.getChannelConnectedTimeout());
            channelSftp = (ChannelSftp) channel;
            oriPath = channelSftp.pwd();
        } catch (Exception e) {
            e.printStackTrace();
            close();
        }
    }

    /**断开连接*/
    public void close() {
        if (channelSftp != null) {
            try {
                channelSftp.disconnect();
            } catch (Exception ignored) {
            }
        }
        if (session != null) {
            try {
                session.disconnect();
            } catch (Exception ignored) {
            }
        }
    }

    /**验证连接*/
    public boolean validate() {
        try {
            return session.isConnected() && channelSftp.isConnected() /*&&
                    !Objects.isNull(oriPath) && oriPath.equals(channelSftp.pwd())*/;
        } catch (Exception e) {
            return false;
        }
    }

    /**
     * 获取当前工作目录
     * */
    public String pwd() throws SftpException {
        return channelSftp.pwd();
    }

    /**获取当前文件的状态*/
    public SftpATTRS stat(String path) throws SftpException {
        return channelSftp.stat(path);
    }

    /**获取当前文件是否是文件夹*/
    public boolean isDir(String path) throws SftpException {
        return channelSftp.stat(path).isDir();
    }

    /**ls*/
    public Vector ls(String path) throws SftpException {
        return channelSftp.ls(path);
    }

    /**mkdir*/
    public void mkdir(String path) throws SftpException {
        channelSftp.mkdir(path);
    }

    /**mkdirs*/
    public void mkdirs(String path) throws SftpException {
        String[] paths = path.split("/");
        for (int i = 0; i < paths.length; i++) {
            StringBuilder pPath = new StringBuilder();
            for (int j = 0; j <= i; j++) {
                pPath.append("/").append(paths[j]);
            }
            String p = pPath.toString();
            p = p.replaceAll("//+", "/");
            if (exist(p)) {
                if (!isDir(p)) {
                    throw new RuntimeException(String.format("路径:[%s]不是路径", p));
                }
            } else {
                if (!"/".equals(p)) {
                    channelSftp.mkdir(p);
                }
            }
        }
    }

    /**touch*/
    public void touch(String path) throws SftpException {
        // todo 未找到类似touch的方式，暂时以上传一个空文件的方式实现
        String parentPath = path.substring(0, path.lastIndexOf("/"));
        if (StringUtils.isEmpty(parentPath)) {
            parentPath = "/";
        }
        String fileName = path.substring(path.lastIndexOf("/") + 1);
        //如果不存在，创建父目录
        mkdirs(parentPath);
        File tmpFile = null;
        try {
            Path tempFile = Files.createTempFile("tmp", "tmp");
            tmpFile = tempFile.toFile();
            try (FileInputStream fis = new FileInputStream(tmpFile)) {
                channelSftp.cd(parentPath);
                channelSftp.put(fis, fileName);
            }
        } catch (SftpException e1) {
            throw e1;
        } catch (Exception e) {
            throw new RuntimeException("创建文件出错", e);
        } finally {
            if (tmpFile != null) {
                tmpFile.delete();
            }
        }
    }

    /**
     * cp
     * 此实现需要该登录用户已被授权ssh功能
     * */
    public void cp(String src, String dest) {
        ChannelExec channelExec = null;
        StringBuilder runLog = new StringBuilder();
        StringBuilder errLog = new StringBuilder();
        try {
            String command = "cp -r " + src + " " + dest;
            Channel channel = session.openChannel("exec");
            channelExec = (ChannelExec) channel;
            channelExec.setCommand(command);
            channelExec.connect();
            // 获取标准输入流
            BufferedReader inputStreamReader = new BufferedReader(new InputStreamReader(channelExec.getInputStream()));
            // 获取标准错误输入流
            BufferedReader errInputStreamReader = new BufferedReader(new InputStreamReader(channelExec.getErrStream()));
            // 记录命令执行 log
            String line;
            while ((line = inputStreamReader.readLine()) != null) {
                runLog.append(line).append("\n");
            }
            //记录命令执行错误 log
            String errLine;
            while ((errLine = errInputStreamReader.readLine()) != null) {
                errLog.append(errLine).append("\n");
            }
            // 输出 shell 命令执行日志
            System.out.println("exitStatus=" + channelExec.getExitStatus() + ", openChannel.isClosed="
                    + channelExec.isClosed());
            System.out.println("命令执行完成，执行日志如下:");
            System.out.println(runLog);
            System.out.println("命令执行完成，执行错误日志如下:");
            System.out.println(errLog);
            if (errLog.length() > 0) {
                throw new RuntimeException(errLog.toString());
            }
        } catch (RuntimeException e) {
           throw e;
        } catch (Exception e) {
            throw new RuntimeException("复制文件出错", e);
        } finally {
            if (channelExec != null) {
                channelExec.disconnect();
            }
        }
    }

    /**
     * mv
     * 此实现需要该登录用户已被授权ssh功能
     * */
    public void mv(String src, String dest) {
        ChannelExec channelExec = null;
        StringBuilder runLog = new StringBuilder();
        StringBuilder errLog = new StringBuilder();
        try {
            String command = "mv " + src + " " + dest;
            Channel channel = session.openChannel("exec");
            channelExec = (ChannelExec) channel;
            channelExec.setCommand(command);
            channelExec.connect();
            // 获取标准输入流
            BufferedReader inputStreamReader = new BufferedReader(new InputStreamReader(channelExec.getInputStream()));
            // 获取标准错误输入流
            BufferedReader errInputStreamReader = new BufferedReader(new InputStreamReader(channelExec.getErrStream()));
            // 记录命令执行 log
            String line;
            while ((line = inputStreamReader.readLine()) != null) {
                runLog.append(line).append("\n");
            }
            //记录命令执行错误 log
            String errLine;
            while ((errLine = errInputStreamReader.readLine()) != null) {
                errLog.append(errLine).append("\n");
            }
            // 输出 shell 命令执行日志
            System.out.println("exitStatus=" + channelExec.getExitStatus() + ", openChannel.isClosed="
                    + channelExec.isClosed());
            System.out.println("命令执行完成，执行日志如下:");
            System.out.println(runLog);
            System.out.println("命令执行完成，执行错误日志如下:");
            System.out.println(errLog);
            if (errLog.length() > 0) {
                throw new RuntimeException(errLog.toString());
            }
        } catch (RuntimeException e) {
            throw e;
        } catch (Exception e) {
            throw new RuntimeException("移动文件出错", e);
        } finally {
            if (channelExec != null) {
                channelExec.disconnect();
            }
        }
    }

    /**
     * 删除文件
     * */
    @SuppressWarnings("unchecked")
    public void rm(String realPath, boolean r) throws SftpException {
        if (isDir(realPath)) {
            // 文件夹，递归
            Vector vector = ls(realPath);
            if (vector.size() <= 0) {
                // 没有子文件或文件夹，直接删除
                channelSftp.rmdir(realPath);
            } else {
                // 如果有子文件或文件夹，通过r参数判定是否能删除
                if (!r) {
                    throw new RuntimeException(String.format("路径:[%s]中存在子文件或文件夹，无法删除", realPath));
                }
                // 递归
                Iterator<ChannelSftp.LsEntry> iterator = vector.iterator();
                while (iterator.hasNext()) {
                    ChannelSftp.LsEntry lsEntry = iterator.next();
                    String filename = lsEntry.getFilename();
                    if (!".".equals(filename) && !"..".equals(filename)) {
                        rm(FsCommonUtils.optimizationPath(realPath + "/" + filename), r);
                    }
                }
                channelSftp.rmdir(realPath);
            }
        } else {
            // 非文件夹直接删除
            channelSftp.rm(realPath);
        }
    }

    /**
     * 查看当前路径或文件否存在
     * */
    public boolean exist(String path) throws SftpException {
        boolean flag = true;
        if ("/".equals(path)) {
            return true;
        }
        try {
            channelSftp.stat(path);
        } catch (SftpException e) {
            if (e.getMessage() != null && e.getMessage().contains("No such file")) {
                flag = false;
            } else {
                throw e;
            }
        }
        return flag;
    }

    /**
     * 上传文件
     * */
    public void uploadFile(String realPath, InputStream is, boolean append) throws SftpException {
        channelSftp.put(is, realPath, append ? ChannelSftp.APPEND : ChannelSftp.OVERWRITE);
    }

    /**
     * 下载文件
     * */
    public void downloadFile(String path, OutputStream os) throws SftpException {
        channelSftp.get(path, os);
    }

    public ChannelSftp getChannelSftp() {
        return channelSftp;
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316

3、连接池属性对象

package cn.ac.iscas.dmo.db.sftp.pool;

import lombok.Data;

/**
 * @author zhuquanwen
 * @version 1.0
 * @date 2022/9/8 9:57
 * @since jdk11
 */
@Data
public class PoolProp {
    /**
     * 池中最小的连接数，只有当 timeBetweenEvictionRuns 为正时才有效
     */
    private int minIdle = 0;

    /**
     * 池中最大的空闲连接数，为负值时表示无限
     */
    private int maxIdle = 8;

    /**
     * 池可以产生的最大对象数，为负值时表示无限
     */
    private int maxActive = 16;

    /**
     * 当池耗尽时，阻塞的最长时间，为负值时无限等待
     */
    private long maxWait = -1;

    /**
     * 从池中取出对象是是否检测可用
     */
    private boolean testOnBorrow = true;

    /**
     * 将对象返还给池时检测是否可用
     */
    private boolean testOnReturn = false;

    /**
     * 检查连接池对象是否可用
     */
    private boolean testWhileIdle = true;

    /**
     * 距离上次空闲线程检测完成多久后再次执行
     */
    private long timeBetweenEvictionRuns = 300000L;


    /**
     * 用户名
     */
    private String username;

    /**
     * 主机
     */
    private String host;

    /**
     * 端口
     */
    private int port;

    /**
     * 密码
     */
    private String password;

    /**
     * session连接超时时间
     */
    private int sessionConnectTimeout = 2000;

    /**
     * sftp-channel连接超时时间
     */
    private int channelConnectedTimeout = 2000;

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

4、连接池

package cn.ac.iscas.dmo.db.sftp.pool;

import cn.ac.iscas.dmo.db.sftp.Sftp;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.ObjectPool;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import java.time.Duration;

/**
 * sftp连接池
 *
 * @author zhuquanwen
 * @version 1.0
 * @date 2022/9/8 9:59
 * @since jdk11
 */
public class SftpPool implements ObjectPool<Sftp> {
    private final GenericObjectPool<Sftp> pool;

    public SftpPool(PoolProp poolProp) {
        pool = new GenericObjectPool<>(new SftpFactory(poolProp), getConfig(poolProp));
    }

    @Override
    public void addObject() throws Exception {
        pool.addObject();
    }

    @Override
    public Sftp borrowObject() throws Exception {
        return pool.borrowObject();
    }

    @Override
    public void clear() {
        pool.clear();
    }

    @Override
    public void close() {
        pool.close();
    }

    @Override
    public int getNumActive() {
        return pool.getNumActive();
    }

    @Override
    public int getNumIdle() {
        return pool.getNumIdle();
    }

    @Override
    public void invalidateObject(Sftp obj) throws Exception {
        pool.invalidateObject(obj);
    }

    @Override
    public void returnObject(Sftp obj) {
        pool.returnObject(obj);
    }

    private static class SftpFactory extends BasePooledObjectFactory<Sftp> {

        private final PoolProp poolProp;

        public SftpFactory(PoolProp poolProp) {
            this.poolProp = poolProp;
        }

        @Override
        public Sftp create() {
            return new Sftp(poolProp);
        }

        @Override
        public PooledObject<Sftp> wrap(Sftp sftp) {
            return new DefaultPooledObject<>(sftp);
        }

        @Override
        public boolean validateObject(PooledObject<Sftp> p) {
            return p.getObject().validate();
        }

        @Override
        public void destroyObject(PooledObject<Sftp> p) {
            p.getObject().close();
        }
    }

    private GenericObjectPoolConfig<Sftp> getConfig(PoolProp poolProp) {
        if (poolProp == null) {
            poolProp = new PoolProp();
        }
        GenericObjectPoolConfig<Sftp> config = new GenericObjectPoolConfig<>();
        config.setMinIdle(poolProp.getMinIdle());
        config.setMaxIdle(poolProp.getMaxIdle());
        config.setMaxTotal(poolProp.getMaxActive());
        config.setMaxWait(Duration.ofMillis(poolProp.getMaxWait()));
        config.setTestOnBorrow(poolProp.isTestOnBorrow());
        config.setTestOnReturn(poolProp.isTestOnReturn());
        config.setTestWhileIdle(poolProp.isTestWhileIdle());
        config.setTimeBetweenEvictionRuns(Duration.ofMillis(poolProp.getTimeBetweenEvictionRuns()));
        return config;
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

5、使用连接池

 PoolProp poolProp = new PoolProp();
//todo 连接池信息先用默认的，这里只设置用户名、密码等信息
poolProp.setUsername("root");
poolProp.setHost("125.2.2.2");
poolProp.setPort(22);
poolProp.setPassword("***");
SftpPool sftpPool = new SftpPool(poolProp);
Sftp sftp = null;
try {
	sftp = sftpPool.borrowObject();
	boolean res = sftp.exist("/root");
} finally {
	sftpPool.returnObject(sftp);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

6、附几个工具类

package cn.ac.iscas.dmo.db.api.util;

import java.util.Objects;

/**
 * @author zhuquanwen
 * @version 1.0
 * @date 2022/3/8 8:52
 * @since jdk11
 */
public class StringUtils {
    private StringUtils() {
    }

    public static boolean isEmpty(Object str) {
        return (str == null || "".equals(str));
    }

    public static boolean isNotEmpty(Object str) {
        return !isEmpty(str);
    }

    public static String substringAfter(final String str, final String separator) {
        if (isEmpty(str)) {
            return str;
        }
        if (separator == null) {
            return "";
        }
        final int pos = str.indexOf(separator);
        if (pos == -1) {
            return "";
        }
        return str.substring(pos + separator.length());
    }

    public static String trySubstringBetween(final String str, final String open, final String close) {
        final int start = str.indexOf(open);
        if (start != -1) {
            final int end = str.indexOf(close, start + open.length());
            if (end != -1) {
                return str.substring(start + open.length(), end);
            }
        }
        return str;
    }

    public static boolean equalsAny(final CharSequence string, final CharSequence... searchStrings) {
        if (!Objects.isNull(searchStrings)) {
            for (final CharSequence next : searchStrings) {
                if (Objects.equals(string, next)) {
                    return true;
                }
            }
        }
        return false;
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

package cn.ac.iscas.dmo.db.filesystem.base.util;

import cn.ac.iscas.dmo.db.api.model.permission.AuthContext;
import cn.ac.iscas.dmo.db.api.model.permission.Datasource;
import cn.ac.iscas.dmo.db.api.model.permission.Permission;
import cn.ac.iscas.dmo.db.api.util.AuthContextHolder;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;

/**
 * @author zhuquanwen
 * @version 1.0
 * @date 2022/6/21 10:44
 * @since jdk11
 */
public class FsCommonUtils {
    private FsCommonUtils() {
    }

    /**
     * 一个KB的字节数
     */
    public static final long KB = 1024L;

    /**
     * 一个MB的字节数
     */
    public static final long MB = 1024 * 1024L;

    /**
     * 一个GB的字节数
     */
    public static final long GB = 1024 * 1024 * 1024L;

    /**
     * 一个TB的字节数
     */
    public static final long TB = 1024 * 1024 * 1024 * 1024L;

    /**
     * 用户目录
     */
    public static final String USER_SPACE_FORMATTER = "/dmo/users/namespace/%s";

    /**
     * URL匹配器
     */
    public static final UrlMatcher URL_MATCHER = new UrlMatcher();

    /**
     * 获取有权限的文件路径
     *
     * @param datasourceName 数据源名称
     * @return java.util.List
     * @date 2022/6/21
     * @since jdk11
     */
    public static List<String> getPathPermission(String datasourceName) {
        return Optional.ofNullable(AuthContextHolder.getContext())
                .map(AuthContext::getPermission)
                .map(Permission::getDatasources).flatMap(ds -> ds.stream().filter(d -> d.getName().equalsIgnoreCase(datasourceName)).findFirst())
                .map(Datasource::getFilePaths)
                .orElse(Collections.emptyList());
    }

    /**
     * 修改路径，将// 替换为/, \替换为/, \\替换为/，使得不同操作系统下的文件分割符统一成 / ,防止拼接时\ / 混用
     *
     * @param path 路径
     * @return java.lang.String 替换后的路径
     * @date 2022/6/21
     * @since jdk11
     */
    public static String optimizationPath(String path) {
        path = path.replaceAll("//+", "/")
                .replace("\\", "/")
                .replace("\\\\", "/");
        if (path.endsWith("/")) {
            path = path.substring(0, path.length() - 1);
        }
        return path;
    }

    /**
     * 格式化文件大小
     *
     * @param len 长度
     * @return java.lang.String
     * @date 2022/6/21
     * @since jdk11
     */
    public static String formatSize(long len) {
        if (len < KB) {
            return len + " B";
        } else if (len < MB) {
            return scale(len * 1.0 / KB, 2) + " KB";
        } else if (len < GB) {
            return scale(len * 1.0 / MB, 2) + " MB";
        } else if (len < TB) {
            return scale(len * 1.0 / GB, 2) + " GB";
        } else {
            return scale(len * 1.0 / TB, 2) + " TB";
        }
    }

    /**
     * 对效数取有效数字
     *
     * @param data  数据
     * @param scale 有效位数
     * @return double
     * @date 2022/6/21
     * @since jdk11
     */
    @SuppressWarnings("SameParameterValue")
    public static double scale(double data, int scale) {
        BigDecimal bg = new BigDecimal(data);
        return bg.setScale(scale, RoundingMode.HALF_UP).doubleValue();
    }

    /**
     * 格式化时间
     *
     * @param timeMillis 毫秒数
     * @return java.lang.String
     * @date 2022/6/21
     * @since jdk11
     */
    public static String formatTime(long timeMillis) {
        return DateSafeUtils.format(new Date(timeMillis), DateSafeUtils.PATTERN);
    }

    /**
     * 获取文件深度
     *
     * @param parentPath 路径
     * @return int
     * @date 2022/6/21
     * @since jdk11
     */
    public static int getLevel(String parentPath) {
        if (parentPath.endsWith("/")) {
            parentPath = parentPath.substring(0, parentPath.lastIndexOf("/"));
        }
        parentPath = optimizationPath(parentPath);
        if (Objects.equals("/", parentPath)) {
            return 1;
        }
        return parentPath.split("/").length;
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155