思享工具箱

commons-pool2 实现 sftp 连接池

摘要：
简介ssh连接的默认数量是有限的。当大量请求连接到ssh时，连接很可能会失败，甚至直接失败。因此，需要合并连接。当然，如果不需要实时，可以使用生产者和消费者。了解commons-pool2依赖项&lt；相关性&gt；&书信电报；组ID&gt；有机沥青。公用&lt；/组ID&gt；&书信电报；artifactId&gt；公用池2&lt；/人工制品

简介

ssh 默认的连接数量有限，当大量请求连接 ssh 时会概率性连接失败甚至直接失败，因此需要对连接池化，当然如果不要求实时的话可以用生产者消费者。

了解 commons-pool2

依赖

<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
    <version>2.8.1</version>
</dependency>

核心概念

  • PooledObjectState：池化对象的状态
    • IDLE：在队列中，未使用。
    • ALLOCATED：正在使用。
    • EVICTION：在队列中，正在判断是否满足被驱逐的条件。
    • EVICTION_RETURN_TO_HEAD：不在队列中，目前正在测试是否可能被驱逐。 测试时尝试借用对象，将其从队列中删除。 一旦
      测试完成，它应该返回到队列的头部。
    • VALIDATION：在队列中，当前正在验证。
    • VALIDATION_PREALLOCATED：预分配状态，不在队列中，还在测试，测试成功后分配。
    • VALIDATION_RETURN_TO_HEAD：不在队列中，目前正在验证。 测试时尝试借用对象，将其从队列中删除。 一旦
      测试完成，它应该返回到队列的头部。
    • INVALID：无效的，将被销毁或已经被销毁。
    • ABANDONED：废弃的
    • RETURNING：使用完毕，正在归还到池中。
  • PooledObject：池化对象的包装器，包含被包装的对象、状态以及使用信息，默认实现是 DefaultPooledObject 。
  • PooledObjectFactory：对象工厂，管理池对象的生命周期（借出、验证和销毁等等），常用的实现是 BasePooledObjectFactory。
  • BaseObjectPoolConfig：对象池配置，实现有 GenericObjectPoolConfig 等等，常用参数如下。
    • minIdle：对象池最小空闲对象数，默认为 0 。
    • maxIdle：对象池最大空闲对象数，默认为 8 。
    • maxTotal：对象池最大可用对象数，默认为 8 。
    • maxWaitMillis：获取对象是最大等待时间，默认为 -1 （负值时为无限等待）。
    • testOnBorrow：从对象池中借出对象时是否进行测试，默认为 false 。
    • testOnReturn：返回给对象池时是否测试对象可用，默认为 false。
    • testWhileIdle：空闲对象驱逐验证开关，打开后定时验证对象池中空闲对象是否可用，不可用就销毁该对象，默认为 false。
    • timeBetweenEvictionRunsMillis：上次空闲对象驱逐验证完成后多久开始下次验证，与 testWhileIdle 配合使用，默认为 -1 （负值时为无限等待）。
  • ObjectPool：对象池。常用的实现是 GenericObjectPool ，它可以配置 PooledObjectFactory （对象工厂）、 GenericObjectPoolConfig （对象池的配置）以及 AbandonedConfig （废弃对象检测，可以不配置）。

简单应用

一个最简单的使用步骤为：

  1. 创建用于包装对象的 PooledObject 。这里使用默认实现 DefaultPooledObject 。
  2. 创建管理对象的工厂 PooledObjectFactory 。新建 CustomPooledObjectFactory ，继承 BasePooledObjectFactory<Object> ，并实现 create() 和 wrap(Object obj) 方法，其中 create() 方法返回的是我们想要放入和借出的对象；wrap(Object obj) 方法返回的是通过 PooledObject 包装的 Object，这里可以使用默认实现 new DefaultPooledObject<Object>(obj) 。
  3. 创建对象池配置。使用常用的 GenericObjectPoolConfig<Object> ，参数通过对应的 set 方法传入。
  4. 创建连接池。使用常用的 GenericObjectPool ，构造函数为 public GenericObjectPool(final PooledObjectFactory<T> factory, final GenericObjectPoolConfig<T> config) ，第一个参数传入第二步创建的 CustomPooledObjectFactory ，第二个参数传入第三步创建的 GenericObjectPoolConfig 。

代码如下，其中 main 方法为创建对象池的 demo 和使用对象池的官方 demo：

import org.apache.commons.pool2.BasePooledObjectFactory;
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;

public class CustomPooledObjectFactory extends BasePooledObjectFactory<Object> {
  @Override
  public Object create() throws Exception {
    return new Object();
  }

  @Override
  public PooledObject<Object> wrap(Object obj) {
    return new DefaultPooledObject<>(obj);
  }

  public static void main(String[] args) {
    final GenericObjectPool<Object> objectPool = new GenericObjectPool<>(new CustomPooledObjectFactory(), new GenericObjectPoolConfig<>());
    Object obj = null;
    try {
      obj = objectPool.borrowObject();
      try {
        // 使用对象
      } catch(Exception e) {
        // 使对象无效
        objectPool.invalidateObject(obj);
        // 不要将对象返回到池中两次
        obj = null;
      } finally {
        // 确保对象返回到池中
        if(null != obj) {
          objectPool.returnObject(obj);
        }
      }
    } catch(Exception e) {
      // 从池中借出对象失败
    }
  }
}

Sftp

sftp 是通过 ssh 完成安全的传输的，而 ssh 的连接数是有限的，具体的 ssh 连接参数见 /etc/ssh/sshd_config 的 MaxStartups 参数，默认为 10 或 10:30:100，含义如下：

  • 10：单独只有一个参数表示最大可用连接数，当 10:30:100 这种格式时表示 100% 可以连接成功的连接数。
  • 30：表示 已经有 10 个连接时再连接有 30% 的概率连接失败。
  • 100：表示最大连接数。

因此为了尽可能少的创建新连接就需要使用连接池。

依赖

<dependency>
    <groupId>com.jcraft</groupId>
    <artifactId>jsch</artifactId>
    <version>0.1.55</version>
</dependency>

创建Sftp连接池

首先创建被池化的对象 SftpClient ，其中 originalDir 属性用于保存登录时的原始目录，validateConnect 方法用于验证连接，disconnect 方法用于销毁连接。

package com.haibara.toys.sftp.core;

import com.jcraft.jsch.*;
import lombok.Data;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author haibara
 */
@Data
public class SftpClient {
  private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
  private static final AtomicLong CLIENT_NUMBER = new AtomicLong(1L);
  private ChannelSftp channelSftp;
  private Session session;
  /**
   *
   */
  private String clientInfo = "sftpclient";
  /**
   * ssh 根目录。
   * 用于判断是否成功返回连接到连接池的条件之一
   */
  private String originalDir;

  public SftpClient(SftpProperties sftpProperties) throws SftpException, JSchException {
    try {
      JSch jsch = new JSch();
      session = jsch.getSession(sftpProperties.getUsername(), sftpProperties.getHost(), sftpProperties.getPort());
      session.setPassword(sftpProperties.getPassword());
      Properties config = new Properties();
      if (sftpProperties.getSession() != null) {
        sftpProperties.getSession().forEach(config::put);
      }
      session.setConfig(config);
      session.connect();
      channelSftp = (ChannelSftp) session.openChannel("sftp");
      channelSftp.connect();
      clientInfo += CLIENT_NUMBER.getAndIncrement() + ",createTime:" + DATE_TIME_FORMATTER.format(LocalDateTime.now());
      originalDir = channelSftp.pwd();
    } catch (Exception e) {
      disconnect();
      throw e;
    }
  }

  public void disconnect() {
    if (channelSftp != null) {
      try {
        channelSftp.disconnect();
      } catch (Exception ignored) {
      }
    }
    if (session != null) {
      try {
        session.disconnect();
      } catch (Exception ignored) {
      }
    }
  }

  public boolean validateConnect() {
    try {
      return session.isConnected() && channelSftp.isConnected() && originalDir.equals(channelSftp.pwd());
    } catch (Exception e) {
      return false;
    }
  }
}

接下来和前面一样分 4 步创建。

  1. 创建被包装的对象，同样使用默认的 DefaultPooledObject 。

  2. 创建对象工厂 SftpFactory ，重写 validateObject 和 destroyObject 方法为 SftpClient 的验证和销毁方法，SftpFactory 代码见第 4 步 SftpPool 的静态内部类 SftpFactory 。

  3. 创建连接池配置（以及 sftp 的配置） SftpProperties 类和 GenericObjectPoolConfig （连接池配置），GenericObjectPoolConfig 创建代码见第 4 步 SftpPool 的 getPoolConfig 方法 。

package com.haibara.toys.sftp.core;

import lombok.Data;

import java.util.Map;

/**
 * @author haibara
 */
@Data
public class SftpProperties {
  /**
   * 地址
   */
  private String host = "localhost";
  /**
   * 端口号
   */
  private int port = 22;
  /**
   * 用户名
   */
  private String username;
  /**
   * 密码
   */
  private String password;
  /**
   * Session 参数配置
   */
  private Map<String, String> session;
  /**
   * 连接池配置
   */
  private Pool pool;

  /**
   * 连接池配置类
   */
  @Data
  public static class Pool {
    /**
     * 池中最小的连接数，只有当 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;
  }
}

yml 中对应的配置如下：

sftp:
  host: 127.0.0.1
  port: 22
  username: root
  password: 123456
  session:
    StrictHostKeyChecking: no
    kex: diffie-hellman-group1-sha1,diffie-hellman-group-exchange-sha1,diffie-hellman-group-exchange-sha256
  pool:
    max-idle: 8
    min-idle: 1
    max-active: 16
    max-wait: 150000
    test-on-borrow: true
    test-on-return: false
    test-while-idle: true
    time-between-eviction-runs: 120000
  1. 创建连接池 SftpPool 。需要注意的一点是返还给连接池的连接要和新连接的状态相同，因此重写 GenericObjectPool 的 returnObject 方法，在原来的 returnObject 前恢复（还原当前目录为初始目录等等），同时在 SftpClient 的 validateConnect 方法中添加判断是否恢复的条件（判断当前目录是否是初始目录等等）。
package com.haibara.toys.sftp.core;

import com.jcraft.jsch.SftpException;
import lombok.SneakyThrows;
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.util.NoSuchElementException;

/**
 * @author haibara
 */
public class SftpPool implements ObjectPool<SftpClient> {
  private final GenericObjectPool<SftpClient> internalPool;

  public SftpPool(SftpProperties sftpProperties) {
    this.internalPool = new GenericObjectPool<SftpClient>(new SftpFactory(sftpProperties), getPoolConfig(sftpProperties.getPool())){
      @Override
      public void returnObject(SftpClient sftpClient) {
        try {
          sftpClient.getChannelSftp().cd(sftpClient.getOriginalDir());
        } catch (Exception ignored) {
        }
        super.returnObject(sftpClient);
      }
    };
  }

  @Override
  public void addObject() throws Exception, IllegalStateException, UnsupportedOperationException {
    internalPool.addObject();
  }

  @Override
  public SftpClient borrowObject() throws Exception, NoSuchElementException, IllegalStateException {
    return internalPool.borrowObject();
  }

  @Override
  public void clear() throws Exception, UnsupportedOperationException {
    internalPool.clear();
  }

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

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

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

  @Override
  public void invalidateObject(SftpClient obj) throws Exception {
    internalPool.invalidateObject(obj);
  }

  @Override
  public void returnObject(SftpClient obj) {
    internalPool.returnObject(obj);
  }

  private static class SftpFactory extends BasePooledObjectFactory<SftpClient> {

    private final SftpProperties sftpProperties;

    public SftpFactory(SftpProperties sftpProperties) {
      this.sftpProperties = sftpProperties;
    }

    @Override
    public SftpClient create() throws Exception {
      return new SftpClient(sftpProperties);
    }

    @Override
    public PooledObject<SftpClient> wrap(SftpClient sftpClient) {
      return new DefaultPooledObject<>(sftpClient);
    }

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

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

  }

  private GenericObjectPoolConfig<SftpClient> getPoolConfig(SftpProperties.Pool properties) {
    if (properties == null) {
      properties = new SftpProperties.Pool();
    }
    GenericObjectPoolConfig<SftpClient> config = new GenericObjectPoolConfig<>();
    config.setMinIdle(properties.getMinIdle());
    config.setMaxIdle(properties.getMaxIdle());
    config.setMaxTotal(properties.getMaxActive());
    config.setMaxWaitMillis(properties.getMaxWait());
    config.setTestOnBorrow(properties.isTestOnBorrow());
    config.setTestOnReturn(properties.isTestOnReturn());
    config.setTestWhileIdle(properties.isTestWhileIdle());
    config.setTimeBetweenEvictionRunsMillis(properties.getTimeBetweenEvictionRuns());
    return config;
  }
}

最后只需要创建一个 SftpPool 对象就好了，使用代码如下（SftpTemplate 是对 SftpClient 封装的工具类）：

@Test
void testSftp() {
  String localFile = "";
  String remotePath = "";
  String fileName = "";
  SftpClient sftpClient = null;
  try (FileInputStream fileInputStream = new FileInputStream(localFile)) {
    try {
      sftpClient = sftpPool.borrowObject();
      SftpTemplate sftpTemplate = new SftpTemplate(sftpClient);
      sftpTemplate.cd(remotePath);
      sftpTemplate.upload(fileInputStream, fileName);
    } catch (Exception e) {
      sftpPool.invalidateObject(sftpClient);
      sftpClient = null;
    } finally {
      if (null != sftpClient) {
        sftpPool.returnObject(sftpClient);
      }
    }
  } catch (Exception e) {
    // 从池中借出对象失败
  }
}

具体代码见 hligaty/Toys 。

参考

apache common pool2原理与实战 - 海向 - 博客园 (cnblogs.com)

标签：#连接池#ftp#string#对象池 | 浏览：697 | 发布日期：

免责声明：文章转载自《commons-pool2 实现 sftp 连接池》仅用于学习参考。如对内容有疑问，请及时联系本站处理。

上篇如何用Apache POI操作Excel文件-----如何在已有的Excel文件中插入一行新的数据？《QT Creator快速入门》第十三章：QT多媒体应用下篇

宿迁高防，2C2G15M，22元/月；香港BGP，2C5G5M，25元/月 雨云优惠码：MjYwNzM=

相关文章

微信开发——使用微信接口（获取地理位置）

生成微信签名（只要访问微信的接口，都需要生成签名验证来进行config） a. 获取AccessToken //定义静态常量存放获取AccessToken的URL public final static String GetPageAccessTokenUrl = "https://api.weixin.qq.com/cgi-bin/token?gra...

dynamic-insert和dynamic-update属性

dynamic-insert 作用:设置对象中没有值的字段 insert并不会对其进行插入. 实体类映射配置如下 <!DOCTYPE hibernate-mapping PUBLIC "-//Hibernate/Hibernate Mapping DTD 3.0//EN" "http://www.hibernate.o...

Drools介绍与使用

Drools 是用 Java 语言编写的开放源码规则引擎，使用 Rete 算法对所编写的规则求值。Drools 允许使用声明方式表达业务逻辑。可以使用非 XML 的本地语言编写规则，从而便于学习和理解。并且，还可以将 Java 代码直接嵌入到规则文件中，这令 Drools 的学习更加吸引人。 Drools 还具有其他优点： 非常活跃的社区支持 易用 快...

FTP命令

（转自：http://www.edu.cn/20010830/210045.shtml） FTP命令是Internet用户使用最频繁的命令之一，不论是在DOS还是UNIX操作系统下使用FTP，都会遇到大量的FTP内部命令。熟悉并灵活应用FTP的内部命令，可以大大方便使用者，并收到事半功倍之效。 　　FTP的命令行格式为：ftp -v -d -i -n -g...

你连Nginx怎么转发给你请求都说不清楚，还好意思说自己不是CRUD工程师？

目录 一、Nginx工作原理二、Nginx进程模型三、Nginx处理HTTP请求流程 Nginx 工作原理 Nginx由内核和模块组成，Nginx本身做的工作实际很少，当它接到一个HTTP请求时，它仅仅是通过查找配置文件将此次请求映射到一个location block。 此location中所配置的各个指令则会启动不同的模块去完成工作，因此模块可以看做Ng...

java过滤特殊字符的正则表达式

// 过滤特殊字符 public staticString StringFilter(String str) throws PatternSyntaxException { // 只允许字母和数字 // String regEx ="[^a-zA-Z0-9]"; // 清除掉所有特殊字符 String regEx="[`~!@#$%^&*()+...

最新文章

随机推荐

思享工具箱导航