行业知识
高吞吐SFTP连接池设计方案
背景
在���代的数据驱动环境中,安全文件传输协议(SFTP)扮演着至关重要的角色,它提供了一种安全、可靠的文件传输方式。我们目前项目是一个大型数据集成平台,跟上下游有很多文件对接是通过SFTP协议,当需要处理大量文件或进行大规模数据迁移时,我们常常会遇到因上下游服务器的限制导致连接访问被中断或者文件传输中断,大大影响系统的文件传输效率。常见的报错例子 com.jcraft.jsch.JSchException: Session.connect: java.net.SocketException: Connection reset、com.jcraft.jsch.JSchException: connection is closed by foreign host、com.jcraft.jsch.JSchException: session is down、com.jcraft.jsch.JSchException: channel is not opened 等。这些连接问题也一直困扰着我们,那我们有没有办法在上下游有限的资源去最大提高文件并行处理能力呢?目前网上的连接池方案只是简单解决了SSH Session复用,但是并没有充分利用同一个SSH Session对象中的SFTP Channel以减少Session的开销,所以当文件量大的时候,还是会因为服务器的SSH Session数量限制从而影响文件并行传输效率。最节省资源又能提高文件并行处理方式的做法应该是同时复用SSH Session以及其SFTP Channel,这样最大并行可以处理的文件数就是 SSH Session数*SFTP Channel数,而不是直接每个文件处理线程开启一个SSH Session,只使用SSH Session的一个Channel。
问题原因
所有上下游的SFTP服务器都是连接限制,一般来说默认情况下,常见的Linux服务器默认MaxSessions是10,MaxStartups是10:30:60。
这里简单对MaxSessions和MaxStartups两个配置参数解释下,这个对我们理解SFTP很重要。
- MaxStartups 10:30:60:这里指服务器开始时可以进行的最大未经身份验证的连接数是10,当超过第一个数字的连接数时(10),SSHD 将开始随机地拒绝新的连接,在达到第三个数字时(60),SSHD 将拒绝所有新的连接。注意,这里是拒绝未经身份认证的连接数,已经登陆的不会限制,也就是客户端如果同时创建一堆连接等待认证的话,等待的数量超过这个配置就会被拒绝。
- MaxSessions 10: 这里是指每个SSH连接可以创建多少个通信通道,例如JSCH中的ChannelSftp文件处理通道。
如果上下游不对其SFTP服务器做任何设置,那么其服务器最大允许同时并行10个SSH连接(超过10个连接服务器SSHD服务就会随机中断连接,服务开始不稳定),每个SSH连接最多可以打开10个SFTP Channel。
这里要说下MaxSessions和MaxStartups参数对Java SFTP客户端的影响
以下以目前常用的SFTP框架JSCH为例子。 MaxStartups 和 MaxSessions 是 SSH 守护进程(SSHD)的配置选项,而 JSch 中的 Session 和 ChannelSftp 是 Java 客户端连接到 SSH 服务的对象。它们之间的关系如下:
- MaxStartups:这是 SSHD 配置的一部分,用于控制并发未经身份验证的连接的数量。这对于防止暴力破解攻击非常有用,因为它限制了同时尝试身份验证的连接数。这个设置对于 JSch 客户端来说,主要影响的是当你尝试并发创建多个 Session 对象(即多个 SSH 连接)时,服务器端可能会开始拒绝额外的连接。
- MaxSessions:这也是 SSHD 配置的一部分,用于限制每个网络连接(即每个 SSH Session)可以打开的最大会话(channel)数量。这个设置对于 JSch 客户端来说,主要影响的是在一个 Session 对象上你可以创建多少个 ChannelSftp 对象(或其他类型的 Channel 对象)。如果你尝试超过 MaxSessions 的限制创建更多的 Channel,服务器端可能会拒绝新的 Channel 创建请求。
总的来说,MaxStartups 和 MaxSessions 是服务器端的限制,它们影响了客户端(例如使用 JSch 的应用程序)可以并行创建多少个 Session 和 Channel。如果我们要设计一个需要处理大量并发 SFTP 传输的系统,需要考虑这些限制,并在必要时调整服务器的 SSHD 配置。
以文件下载为例,这里用现实中的模型简单说下目前我们数据集成平台下载文件的场景
我们数据集成平台需要从不同的上游的下载文件,A、B、C公司相当于我们的上游系统,某东快递相当于我们的数据集成系统。如下图所示,以其中一个快递公司的视角,某东快递需要对接多个商家收件,如果想提高收件速度,最理想就是安排不同的员工并行到不同的商家去取件,这样才可以同时最快收件;而以其中一个商家的视角,A公司要应对不同快递公司的并行取件请求,必然需要对人员有些限制,不然快递员太多可能会影响公司的正常运作,这也就是为什么SFTP服务器需要限制外部连接。
这里用现实生活中的场景来类比我们数据集成平台和上游系统进行文件下载交互的模型,方便大家理解SFTP服务器的限制,以及面对上游服务器的限制怎么最大化我们的传输效率。P.S:以下内容是以我们数据集成平台的视角去陈述。
1对1模型(对接一个上游下载文件)
因为我们目的是提升我们数据集成平台的并行文件传输效率,所以这里是以我们数据集成平台为参照物,假设只跟一个上游交互的场景。(这里为了简单化说明,假设上游服务器每个用户允许最大并发连接是1,打开最大会话数量是2)。
用例说明: 某东快递类比我们的数据集成平台,我们平台要去上游收件,而A公司类比我们要对接的其中一个上游。A公司的门禁类比上游服务器的限制规则。某东快递的快递员类比数据集成平台需要创建的SSH连接,每个快递员可以同时搬运几个货物类比每个SSH连接打开的SFTP Channel数量。
场景:
某东快递需要上门到A公司取件,A公司有三个件在仓库,A公司有以下门禁要求,不满足就会被拒绝访问:
1.同一个时间允许同一家物流公司的3个快递员进入
2.每个工人不能取超过一个货物
某东快递如果想取件时效性高,在快递员充足的情况下,最理想是派3个快递员同时取件,每人拿一个就可以一次拿完。但是由于A公司有门禁要求,3名快递员同时取取件,会有一位快递员会被保安拒绝,造成人员浪费。因此只需要同时派2名快递员,一个人拿2个,另一个人拿一个就可以。这样子就能一次拿完,也不会人员浪费,时效性也高。
但是真实情况,在对接上游时,对方一般都不会告诉我们有什么限制,通常都是被对方拦截才知道,而且快递员也是有成本的。所以我们作为某东快递,如果一直派快递员去A公司取件,除了可能派出去的快递员会被拦截,还会造成人员浪费。因此我们需要一种灵活配置的方式,可以基于对接的上游,灵活分配符合需求的快递员,这样可以在符合上游限制范围内,最大化我们的取件效率。
我们需要有一种可配置的池化技术,除了可以重复利用快递员,还需要在对方限制的范围基于以下原则合理分配:
- 并行派出的快递员数 private ConcurrentHashMap this.sessions = new ConcurrentHashMap String sessionKey = host + ":" + port + ":" + username; if (StringUtils.hasLength(lockKey)) { // waiting for the available lock while (!lock.lock(lockKey)) { Thread.sleep(1000); // wait a bit before retrying log.debug("Thread {} failed to get lock ", Thread.currentThread().getId()); } log.debug("Thread {} get lock successfully", Thread.currentThread().getId()); } try { return getSessionFromPool(host, port, username, password, timeout, unit, sessionKey); } finally { if (StringUtils.hasLength(lockKey)) { lock.unlock(lockKey); log.debug("Thread {} release lock successfully", Thread.currentThread().getId()); } } } private Session getSessionFromPool(String host, int port, String username, String password, long timeout, TimeUnit unit, String sessionKey) throws JSchException, InterruptedException { ConcurrentLinkedQueue for (Session session : hostSessions) { if (!session.isConnected()) { // This session is no longer valid, remove it from the pool log.warn("This session is no longer valid, remove it from the pool ,sessionId: {}, session detail: {}", session, sessionKey); hostSessions.remove(session); channelCounts.remove(session); maxSessionsPerHostSemaphore.release(); continue; } Semaphore channelSemaphore = channelCounts.get(session); if (channelSemaphore != null && channelSemaphore.tryAcquire()) { return session; } } if (maxSessionsPerHostSemaphore.tryAcquire()) { try { Session session = createNewSession(host, port, username, password); hostSessions.add(session); channelCounts.put(session, new Semaphore(maxChannelsPerSession - 1)); // one channel is already in use return session; } catch (JSchException e) { maxSessionsPerHostSemaphore.release(); throw e; } } Thread.sleep(100); // wait a bit before retrying } throw new NoAvailableSessionException("Timeout while waiting for a session"); } private Session createNewSession(String host, int port, String username, String password) throws JSchException { JSch jsch = new JSch(); Session session = jsch.getSession(username, host, port); session.setPassword(password); session.setConfig("StrictHostKeyChecking", "no"); // Enable StrictHostKeyChecking session.setTimeout(5000); // Set connection timeout to 5 seconds session.connect(); return session; } public ChannelSftp getChannel(Session session) throws JSchException { try { return (ChannelSftp) session.openChannel("sftp"); } catch (JSchException e) { if (!session.isConnected()) { // The session is no longer valid, remove it from the pool String key = session.getHost() + ":" + session.getUserName(); ConcurrentLinkedQueue Semaphore channelSemaphore = channelCounts.get(session); if (channelSemaphore != null) { channelSemaphore.release(); if (channelSemaphore.availablePermits()
SFTP_CONFIG表
这里我创建了一张配置表,用来配置连接池的上限参数,这里JSCH的 MaxSession对应的是服务器MaxStartups参数,JSCH的MaxChannel对应的是服务器的MaxSessions参数
CREATE TABLE SFTP_CONFIG ( ID INT AUTO_INCREMENT PRIMARY KEY, HOST VARCHAR(255), PORT INT, USERNAME VARCHAR(255), PASSWORD VARCHAR(255), MAXSESSIONS INT, MAXCHANNELS INT);
测试连接池
1.不超过服务器限制
1.1 这里我把SFTP服务器的MaxStartups设置成2,MaxSession设置成5。这里把SFTP 的限制设低点是方便测试。
parallels@ubuntu-linux-22-04-desktop:~$ grep -i "MaxStartups" /etc/ssh/sshd_configMaxStartups 2parallels@ubuntu-linux-22-04-desktop:~$ grep -i "MaxSessions" /etc/ssh/sshd_configMaxSessions 5parallels@ubuntu-linux-22-04-desktop:~$
1.2 把连接池的数据库配置改成MaxSession=3,MaxChannel =5 与上面服务器配置一致。
INSERT INTO SFTP_CONFIG (HOST, PORT, USERNAME, PASSWORD, MAXSESSIONS, MAXCHANNELS) VALUES('192.168.50.58', 22,'parallels', 'test8808', 2, 5);1.3 接下来我们来启动程序,会同时并行开启10个线程持有相同的连接信息去访问同一台SFTP 服务器,我们看看程序是否会符合我们预期,10个线程应该只会打开2个SFTP Session,并且连接服务器不会报错。
2024-03-09 01:22:03.536 INFO 58516 --- [ Thread-3] pool.demo.SftpFileProcessThread : Thread 35 got session com.jcraft.jsch.Session@5cae46b5. Session detail: 192.168.50.58:parallels2024-03-09 01:22:04.072 INFO 58516 --- [ Thread-4] pool.demo.SftpFileProcessThread : Thread 36 got session com.jcraft.jsch.Session@5cae46b5. Session detail: 192.168.50.58:parallels2024-03-09 01:22:04.072 INFO 58516 --- [ Thread-7] pool.demo.SftpFileProcessThread : Thread 39 got session com.jcraft.jsch.Session@5cae46b5. Session detail: 192.168.50.58:parallels2024-03-09 01:22:05.075 INFO 58516 --- [ Thread-8] pool.demo.SftpFileProcessThread : Thread 40 got session com.jcraft.jsch.Session@5cae46b5. Session detail: 192.168.50.58:parallels2024-03-09 01:22:05.075 INFO 58516 --- [ Thread-6] pool.demo.SftpFileProcessThread : Thread 38 got session com.jcraft.jsch.Session@5cae46b5. Session detail: 192.168.50.58:parallels2024-03-09 01:22:05.191 INFO 58516 --- [ Thread-9] pool.demo.SftpFileProcessThread : Thread 41 got session com.jcraft.jsch.Session@17e2aaef. Session detail: 192.168.50.58:parallels2024-03-09 01:22:06.083 INFO 58516 --- [ Thread-11] pool.demo.SftpFileProcessThread : Thread 43 got session com.jcraft.jsch.Session@17e2aaef. Session detail: 192.168.50.58:parallels2024-03-09 01:22:06.084 INFO 58516 --- [ Thread-5] pool.demo.SftpFileProcessThread : Thread 37 got session com.jcraft.jsch.Session@17e2aaef. Session detail: 192.168.50.58:parallels2024-03-09 01:22:06.086 INFO 58516 --- [ Thread-12] pool.demo.SftpFileProcessThread : Thread 44 got session com.jcraft.jsch.Session@17e2aaef. Session detail: 192.168.50.58:parallels2024-03-09 01:22:07.084 INFO 58516 --- [ Thread-10] pool.demo.SftpFileProcessThread : Thread 42 got session com.jcraft.jsch.Session@17e2aaef. Session detail: 192.168.50.58:parallels2024-03-09 01:24:03.692 INFO 58516 --- [ Thread-3] pool.demo.SftpFileProcessThread : Thread 35 closed session com.jcraft.jsch.Session@5cae46b5. Session detail: 192.168.50.58:parallels2024-03-09 01:24:04.157 INFO 58516 --- [ Thread-4] pool.demo.SftpFileProcessThread : Thread 36 closed session com.jcraft.jsch.Session@5cae46b5. Session detail: 192.168.50.58:parallels2024-03-09 01:24:04.160 INFO 58516 --- [ Thread-7] pool.demo.SftpFileProcessThread : Thread 39 closed session com.jcraft.jsch.Session@5cae46b5. Session detail: 192.168.50.58:parallels2024-03-09 01:24:05.169 INFO 58516 --- [ Thread-8] pool.demo.SftpFileProcessThread : Thread 40 closed session com.jcraft.jsch.Session@5cae46b5. Session detail: 192.168.50.58:parallels2024-03-09 01:24:05.198 INFO 58516 --- [ Thread-6] pool.demo.SftpFileProcessThread : Thread 38 closed session com.jcraft.jsch.Session@5cae46b5. Session detail: 192.168.50.58:parallels2024-03-09 01:24:05.298 INFO 58516 --- [ Thread-9] pool.demo.SftpFileProcessThread : Thread 41 closed session com.jcraft.jsch.Session@17e2aaef. Session detail: 192.168.50.58:parallels2024-03-09 01:24:06.176 INFO 58516 --- [ Thread-11] pool.demo.SftpFileProcessThread : Thread 43 closed session com.jcraft.jsch.Session@17e2aaef. Session detail: 192.168.50.58:parallels2024-03-09 01:24:06.182 INFO 58516 --- [ Thread-5] pool.demo.SftpFileProcessThread : Thread 37 closed session com.jcraft.jsch.Session@17e2aaef. Session detail: 192.168.50.58:parallels2024-03-09 01:24:06.182 INFO 58516 --- [ Thread-12] pool.demo.SftpFileProcessThread : Thread 44 closed session com.jcraft.jsch.Session@17e2aaef. Session detail: 192.168.50.58:parallels2024-03-09 01:24:07.185 INFO 58516 --- [ Thread-10] pool.demo.SftpFileProcessThread : Thread 42 closed session com.jcraft.jsch.Session@17e2aaef. Session detail: 192.168.50.58:parallels
从以上日志可以看出,10条线程虽然都并行持有相同的SFTP连接信息请求连接池,但是只会获取到2个SFTP Session实例而不是10个SFTP Session,并且当连接池里的SSH Session持有的Channel还没有达到配置的上限5个时,只会分配同一个SSH Session实例,当前连接池里的SSH Session对象持有的Channel超过5才会分配下一个SFTP Session对象,来确保SFTP Session不会浪费。因此目前连接池是符合我们预期想要的效果。
2.超过服务器MaxSessions
2.1 服务器配置跟上面一样,MaxStartups设置成2,MaxSession设置成5。
parallels@ubuntu-linux-22-04-desktop:~$ grep -i "MaxStartups" /etc/ssh/sshd_configMaxStartups 2parallels@ubuntu-linux-22-04-desktop:~$ grep -i "MaxSessions" /etc/ssh/sshd_configMaxSessions 5parallels@ubuntu-linux-22-04-desktop:~$
2.2 把连接池的数据库配置改成MaxSession=3,MaxChannel =1 ,目的是测试当连接池配置的最大Channel数大于服务器限制时,程序多个线程并行进行文件处理会不会报错。
INSERT INTO SFTP_CONFIG (HOST, PORT, USERNAME, PASSWORD, MAXSESSIONS, MAXCHANNELS) VALUES('192.168.50.57', 22,'parallels', 'test8808', 1, 6);2.3 接下来我们来启动程序,会同时并行开启6个线程持有相同的连接信息去访问同一台SFTP 服务器,我们看看程序是否会符合我们预期。如果程序符合预期的话,6个线程会同时使用1个SSH Session去尝试打开6个Channel,第6个线程开始打开Channel进行文件处理的时候应该会开始报错,因为已经超过了服务器所能承受的 MaxSession=5。
2024-03-09 15:30:16.725 INFO 74635 --- [ main] pool.services.SftpService : config->SftpConfig(id=1, host=192.168.50.57, port=22, username=parallels, password=test8808, maxSessions=1, maxChannels=6)2024-03-09 15:30:17.180 INFO 74635 --- [ Thread-6] pool.demo.SftpFileProcessThread : Thread 38 got session com.jcraft.jsch.Session@63c4e36c. Session detail: 192.168.50.57:parallels2024-03-09 15:30:17.257 INFO 74635 --- [ Thread-3] pool.demo.SftpFileProcessThread : Thread 35 got session com.jcraft.jsch.Session@63c4e36c. Session detail: 192.168.50.57:parallels2024-03-09 15:30:17.257 INFO 74635 --- [ Thread-5] pool.demo.SftpFileProcessThread : Thread 37 got session com.jcraft.jsch.Session@63c4e36c. Session detail: 192.168.50.57:parallels2024-03-09 15:30:17.259 INFO 74635 --- [ Thread-8] pool.demo.SftpFileProcessThread : Thread 40 got session com.jcraft.jsch.Session@63c4e36c. Session detail: 192.168.50.57:parallels2024-03-09 15:30:17.264 INFO 74635 --- [ Thread-7] pool.demo.SftpFileProcessThread : Thread 39 got session com.jcraft.jsch.Session@63c4e36c. Session detail: 192.168.50.57:parallels2024-03-09 15:30:17.268 INFO 74635 --- [ Thread-4] pool.demo.SftpFileProcessThread : Thread 36 got session com.jcraft.jsch.Session@63c4e36c. Session detail: 192.168.50.57:parallels2024-03-09 15:30:17.278 ERROR 74635 --- [ Thread-4] pool.demo.SftpFileProcessThread : Thread 36 failed to processcom.jcraft.jsch.JSchException: channel is not opened.at com.jcraft.jsch.Channel.sendChannelOpen(Channel.java:835) ~[jsch-0.2.16.jar:0.2.16]at com.jcraft.jsch.Channel.connect(Channel.java:161) ~[jsch-0.2.16.jar:0.2.16]at com.jcraft.jsch.Channel.connect(Channel.java:155) ~[jsch-0.2.16.jar:0.2.16]at pool.demo.SftpFileProcessThread.run(SftpFileProcessThread.java:54) ~[classes/:na]
从上面日志可以看到,当超过服务器最大Channel配置5时,服务器就会开始拒绝请求,符合我们测试预期。因此我们连接池的配置范围一定要小于等于服务器的配置,否则可能会引起文件传输中断,也就是宁愿少配也不要多配。
3.超过服务器MaxStartups
3.1 服务器配置跟上面一样,MaxStartups设置成2,MaxSession设置成5。
parallels@ubuntu-linux-22-04-desktop:~$ grep -i "MaxStartups" /etc/ssh/sshd_configMaxStartups 2parallels@ubuntu-linux-22-04-desktop:~$ grep -i "MaxSessions" /etc/ssh/sshd_configMaxSessions 5parallels@ubuntu-linux-22-04-desktop:~$
3.2 把连接池的数据库配置改成MaxSession=3,MaxChannel =1 ,目的是测试当连接池配置的最大Session数大于服务器限制时,程序多个线程并发创建SSH Session是否会出错。
INSERT INTO SFTP_CONFIG (HOST, PORT, USERNAME, PASSWORD, MAXSESSIONS, MAXCHANNELS) VALUES('192.168.50.57', 22,'parallels', 'test8808', 3, 1);3.3 修改SftpFileProcessThread代码中获取连接方式改成无锁方式,这样可以看到并发创建Session数量超过服务器限制时服务器的报错。因为MaxStartups上面我们配置了2,需要同时多个线程模拟并发创建超过2个等待验证的Session才能看到效果,如果使用有锁模式,创建Session就会变成串行,验证通过才会到下一个Session,服务器就不会同时有多个等待的验证的Session。
session = pool.getSession(host, port, username, password, 10, TimeUnit.SECONDS, ""); //session = pool.getSession(host, port, username, password, 10, TimeUnit.SECONDS, host + ":" + username); //get session with lock
3.4 接下来我们来启动程序,会同时并行开启3个线程持有相同的连接信息去访问同一台SFTP 服务器,我们看看程序是否会符合我们预期。如果程序符合预期的话,3个线程会同时创建3个SSH Session,第3个线程尝试创建SSH Session 时应该会报错,因为超过了服务器最大 并发Session数限制MaxStartups=2。
2024-03-09 15:55:23.367 INFO 7577 --- [ main] pool.services.SftpService : config->SftpConfig(id=1, host=192.168.50.57, port=22, username=parallels, password=test8808, maxSessions=3, maxChannels=1)2024-03-09 15:55:23.455 ERROR 7577 --- [ Thread-5] pool.demo.SftpFileProcessThread : Thread 37 failed to processcom.jcraft.jsch.JSchException: Session.connect: java.net.SocketException: Connection resetat com.jcraft.jsch.Session.connect(Session.java:570) ~[jsch-0.2.16.jar:0.2.16]at com.jcraft.jsch.Session.connect(Session.java:199) ~[jsch-0.2.16.jar:0.2.16]at pool.common.utils.SftpSessionPool.createNewSession(SftpSessionPool.java:132) ~[classes/:na]at pool.common.utils.SftpSessionPool.getSessionFromPool(SftpSessionPool.java:109) ~[classes/:na]at pool.common.utils.SftpSessionPool.getSession(SftpSessionPool.java:57) ~[classes/:na]at pool.demo.SftpFileProcessThread.run(SftpFileProcessThread.java:39) ~[classes/:na]Caused by: java.net.SocketException: Connection resetat java.net.SocketInputStream.read(SocketInputStream.java:210) ~[na:1.8.0_231]at java.net.SocketInputStream.read(SocketInputStream.java:141) ~[na:1.8.0_231]at java.net.SocketInputStream.read(SocketInputStream.java:224) ~[na:1.8.0_231]at com.jcraft.jsch.IO.getByte(IO.java:84) ~[jsch-0.2.16.jar:0.2.16]at com.jcraft.jsch.Session.connect(Session.java:276) ~[jsch-0.2.16.jar:0.2.16]... 5 common frames omitted2024-03-09 15:55:23.900 INFO 7577 --- [ Thread-3] pool.demo.SftpFileProcessThread : Thread 35 got session com.jcraft.jsch.Session@4777f0b1. Session detail: 192.168.50.57:parallels2024-03-09 15:55:23.900 INFO 7577 --- [ Thread-4] pool.demo.SftpFileProcessThread : Thread 36 got session com.jcraft.jsch.Session@326d5e63. Session detail: 192.168.50.57:parallels
从上面日志可以看到,3个线程同一时间只有2个线程能成功创建Session,另外一个线程创建Session就报错,这是因为超过了服务器最大并发Session数限制而被拒绝访问。
总结
我们如果使用这种可配置连接池进行访问,对接上游时最好时最好跟上游确认他们服务器可以承受的Session数量和Channel数量是多少,宁愿少配也不要多配。但是对于一般上游,如果使用的是Linux服务器,默认值就是上面一开始提到的MaxSessions=10,MaxStartups=10:30:60,我们客户端连接池保守配置可以设置成MaxSession=3,MaxChannel=10,这样子最大并行可以处理30个文件,MaxSession一般不建议设置太多,非常消耗系统资源。也有很多上游是使用其他SFTP服务管理工具,但是基本限制参数差不多。以上连接池还有很多需要完善的地方,大家可以根据需要自己进行优化。
Github源代码
https://github.com/EvanLeung08/sftp-session-pool
- 并行派出的快递员数 private ConcurrentHashMap this.sessions = new ConcurrentHashMap String sessionKey = host + ":" + port + ":" + username; if (StringUtils.hasLength(lockKey)) { // waiting for the available lock while (!lock.lock(lockKey)) { Thread.sleep(1000); // wait a bit before retrying log.debug("Thread {} failed to get lock ", Thread.currentThread().getId()); } log.debug("Thread {} get lock successfully", Thread.currentThread().getId()); } try { return getSessionFromPool(host, port, username, password, timeout, unit, sessionKey); } finally { if (StringUtils.hasLength(lockKey)) { lock.unlock(lockKey); log.debug("Thread {} release lock successfully", Thread.currentThread().getId()); } } } private Session getSessionFromPool(String host, int port, String username, String password, long timeout, TimeUnit unit, String sessionKey) throws JSchException, InterruptedException { ConcurrentLinkedQueue for (Session session : hostSessions) { if (!session.isConnected()) { // This session is no longer valid, remove it from the pool log.warn("This session is no longer valid, remove it from the pool ,sessionId: {}, session detail: {}", session, sessionKey); hostSessions.remove(session); channelCounts.remove(session); maxSessionsPerHostSemaphore.release(); continue; } Semaphore channelSemaphore = channelCounts.get(session); if (channelSemaphore != null && channelSemaphore.tryAcquire()) { return session; } } if (maxSessionsPerHostSemaphore.tryAcquire()) { try { Session session = createNewSession(host, port, username, password); hostSessions.add(session); channelCounts.put(session, new Semaphore(maxChannelsPerSession - 1)); // one channel is already in use return session; } catch (JSchException e) { maxSessionsPerHostSemaphore.release(); throw e; } } Thread.sleep(100); // wait a bit before retrying } throw new NoAvailableSessionException("Timeout while waiting for a session"); } private Session createNewSession(String host, int port, String username, String password) throws JSchException { JSch jsch = new JSch(); Session session = jsch.getSession(username, host, port); session.setPassword(password); session.setConfig("StrictHostKeyChecking", "no"); // Enable StrictHostKeyChecking session.setTimeout(5000); // Set connection timeout to 5 seconds session.connect(); return session; } public ChannelSftp getChannel(Session session) throws JSchException { try { return (ChannelSftp) session.openChannel("sftp"); } catch (JSchException e) { if (!session.isConnected()) { // The session is no longer valid, remove it from the pool String key = session.getHost() + ":" + session.getUserName(); ConcurrentLinkedQueue Semaphore channelSemaphore = channelCounts.get(session); if (channelSemaphore != null) { channelSemaphore.release(); if (channelSemaphore.availablePermits()
上一篇:独立服务器哪里买便宜?
下一篇:怎么进icu服务器?