Java实现_ssh远程会话连接池实现_使用ObjectPool和PooledObjectFactory

一、需求背景

    公司的大数据集群作为基础平台，为公司内部各应用提供计算和存储能力，为实现各应用单独管理并进行资源隔离，一般采用多租户管理。集群为应用租户分配了固定的计算资源，如下应用租户B，应用端在利用spark连接大数据集群时，会根据executor参数在yarn上发布常驻Application进程，锁定相应的计算资源。应用内部可通过不同的Application拆分不同计算资源队列，为不同的机构提供统一的平台能力的同时，各自的计算任务又相互不影响，实现多租户资源租赁模式。

    应用端不同机构的计算资源需要监控起来，方便资源划分和调整，而不同机构的计算服务部署在不同的机器上，所以需要远程连接到不同的服务器进行配置文件的解析和读取，读取的参数如下。这些参数可以根据不同的时间段进行调整，区分上班时间段和休息时间段，不过这样需要依赖重启来锁定不同的计算资源。

totalExecutorCores总核数executorMemory单executor内存executorCores单executor核数executorNumexecutor数量driverMemory服务器驱动内存

二、远程连接并执行linux命令

    java通过jsch包远程执行linux命令，jsch是ssh2的一个纯java实现，可以通过代码连接到一台sshd服务器。jsch进行服务器连接时首先需要实例化一个jsch对象，然后利用这个对象根据用户名，主机ip，端口获取一个Session对象，设置好相应的参数(如认证**公钥路径**或者**密码**)后进行连接，创建连接后这个session时一直可用的，所以不需要关闭，下面优化会使用到对象连接池。最后在session上建立channel通道，channel通道有以下几种，这次主要使用ChannelExec通道来执行远程命令。

（1）对于ChannelShell，以输入流的形式提供命令并输入这些命令，类似于本地计算机上使用交互式shell，多用于交互式；

（2）对于ChannelExec，在调用connect()方法之前需要调用setCommand()方法，并且这些命令将以输入流的形式发送出去。通常只能调用setCommand()方法一次，多次调用只有最后一次生效，但是可以使用普通shell的分隔符(&, &&, |, ||, ; , \n来分割命令)来提供多个命令；

（3）对于ChannelSftp，该对象实现文件上传下载，ChannelSftp类是Jsch实现SFTP核心类，包含了所有SFTP的方法。

2.1 环境准备

    生产上的服务器用户直接利用密码进行远程连接是不安全的，这里通过公钥进行连接，生成和配置公私钥的步骤如下，与linux服务器间配置免密的方法类似。linux免密登录，本质上是使用了”公钥登录”，就是用户将自己的公钥储存在远程主机上，a)在登录的时候，远程主机会向用户发送一段随机字符串，b)用户用自己的私钥加密后再发给远程主机，c)远程主机用事先储存的公钥进行解密，如果成功就证明用户是可信的，直接允许登录shell，不再要求密码。

#在本机生成密钥对
ssh-keygen -t rsa 
#(连续三次回车，即在本地生成了公钥和私钥，不设置密码，默认存储在 ~/.ssh目录下)
 
#将公钥id_rsa.pub追加到需要远程登录的服务端 ~/.ssh/authorized_keys文件中
#这里默认已经将公钥上传到了服务端
cat id_rsa.pub >> ~/.ssh/authorized_keys
chmod 700 ~/.ssh
chmod 600 ~/.ssh/authorized_keys
 
 
#将私钥id_rsa放到客户端~/.ssh/目录下
#这里默认已经将私钥上传到了客户端
mv id_rsa ~/.ssh/
chmod 700 ~/.ssh
chmod 600 ~/.ssh/id_rsa

    **客户端使用私钥，服务端使用公钥**。windows本机调试可以使用Scrt工具生成公私钥，将id_rsa.pub放入需要远程登录服务器的authorized_keys中，代码中使用id_rsa私钥进行远程登录。

2.2 具体代码实现

（1）引入配置包

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

（2）定义配置实体

import lombok.Data;

@Data
public class SshClientConfig {

    private String host; //主机
    private int port;//端口
    private String user;//登录用户
    private String pwd;//登录密码
    private String rsaPath;//登录密钥路径
    private String command;//执行命令
    private String encoding;//编码
    
}

（2）编写ssh连接工具类

import com.jcraft.jsch.ChannelExec;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;
import org.apache.commons.lang3.StringUtils;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;

/**
 * ssh远程连接执行命令工具类
 */
public class SshClientUtil {

    /**
     * 建立连接并获取会话
     * @param config
     * @return
     */
    public static Session connect(SshClientConfig config){
        Session session = null;

        try {
            JSch jSch = new JSch();
            if (!StringUtils.isEmpty(config.getRsaPath())){
                jSch.addIdentity(config.getRsaPath());
            }
            session = jSch.getSession(config.getUser(),config.getHost(),config.getPort());
            if (!StringUtils.isEmpty(config.getPwd())){
                session.setPassword(config.getPwd());
            }
            session.setConfig("StrictHostKeyChecking", "no");
            session.connect();
        } catch (JSchException e) {
            e.printStackTrace();
        } finally {
            if (session != null){
                session.disconnect();
            }
        }
        return session;
    }

    /**
     * 执行远程命令并返回结果
     * @param config
     * @return
     */
    public static List<String> exeCmd(SshClientConfig config){
        Session session = connect(config);
        ChannelExec channelExec = null;
        BufferedReader bufferedReader = null;
        List<String> result = new ArrayList<>();
        try {
            channelExec = (ChannelExec) session.openChannel("exec");
            channelExec.setCommand(config.getCommand());
            channelExec.connect();
            bufferedReader = new BufferedReader(new InputStreamReader(channelExec.getInputStream(), config.getEncoding()));
            String str;
            while(!StringUtils.isEmpty(str = bufferedReader.readLine())){
                result.add(str);
            }
        } catch (JSchException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            if (bufferedReader != null){
                try {
                    bufferedReader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (channelExec != null){
                channelExec.disconnect();
            }
            if (session != null){
                session.disconnect();
            }
        }
        return result;
    }

}

三、对象连接池封装远程连接会话

    com.jcraft.jsch.Session对象的创建开销是比较大的，创建和关闭也比较耗时。为了减少频繁创建、销毁对象带来的性能消耗，可以利用对象池的技术来实现对象的复用。对象池提供了一种机制，它可以管理对象池中对象的生命周期，提供了获取和释放对象的方法，可以让客户端很方便的使用对象池中的对象。对象池一般要完成如下功能：

（1）如果池中有可用的对象，对象池应当能返回给客户端
（2）客户端把对象放回池里后，可以对这些对象进行重用
（3） 对象池能够创建新的对象来满足客户端不断增长的需求
（4）需要有一个正确关闭池的机制来结束对象的生命周期

    Apache 提供的common-pool工具包，里面包含了开发通用对象池的一些接口和实现类，其中最基本的两个接口是ObjectPool 和PooledObjectFactory。

3.1 实现思路

    由于该功能需要连接多台服务器，一台服务器需要建立一个连接池，故使用一个map来存储对应host的连接池，同时利用同步锁保障一个host只能新建一个ssh对象池，各类的依赖情况如下，由SshClientUtil统一对外提供工具接口，SshClientConfig实体设置ssh连接参数和连接池配置实体GenericObjectPoolConfig。

3.2 具体代码实现

（1）引入配置包

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

（2）SshClientFactory.java

import com.jcraft.jsch.JSch;
import com.jcraft.jsch.Session;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;

/**
 * ssh连接对象创建/销毁工厂
 */
public class SshClientFactory implements PooledObjectFactory<Session>{

    private SshClientConfig config;

    public SshClientFactory(SshClientConfig sshClientConfig){
        this.config = sshClientConfig;
    }

    @Override
    public PooledObject<Session> makeObject() throws Exception {
        JSch jSch = new JSch();

        if (!StringUtils.isEmpty(config.getRsaPath())){
            jSch.addIdentity(config.getRsaPath());
        }
        Session session = jSch.getSession(config.getUser(),config.getHost(),config.getPort());
        if (!StringUtils.isEmpty(config.getPwd())){
            session.setPassword(config.getPwd());
        }
        session.setConfig("StrictHostKeyChecking", "no");
        session.connect();

        return new DefaultPooledObject<>(session);
    }

    @Override
    public void destroyObject(PooledObject<Session> pooledObject) throws Exception {
        if (pooledObject.getObject() != null){
            pooledObject.getObject().disconnect();
        }
    }

    @Override
    public boolean validateObject(PooledObject<Session> pooledObject) {
        return pooledObject.getObject().isConnected();
    }

    @Override
    public void activateObject(PooledObject<Session> pooledObject) throws Exception {

    }

    @Override
    public void passivateObject(PooledObject<Session> pooledObject) throws Exception {

    }
}

（3）SshClientPool.java

import com.jcraft.jsch.Session;
import org.apache.commons.pool2.impl.GenericObjectPool;

/**
 * ssh连接对象池
 */
public class SshClientPool {

    private GenericObjectPool<Session> pool;

    public SshClientPool(SshClientConfig sshClientConfig){
        init(sshClientConfig);
    }

    public void init(SshClientConfig sshClientConfig){
        SshClientFactory factory = new SshClientFactory(sshClientConfig);
        pool = new GenericObjectPool<>(factory, sshClientConfig.getObjectPoolConfig());
    }

    /**
     * 获取会话
     * @return
     */
    public Session getSession(){
        try{
            return pool.borrowObject();
        }catch (Exception e){
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 归还会话
     * @param session
     */
    public void releaseSession(Session session){
        try {
            if (session != null){
                pool.returnObject(session);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

（3）SshClientHostPool.java

import java.util.HashMap;
import java.util.Map;

/**
 * 不同的主机对应不同的ssh连接池
 */
public class SshClientHostPool {

    public static Map<String, SshClientPool> poolMap = new HashMap<>();

    public static SshClientPool getHostPool(SshClientConfig config){

        if (!poolMap.containsKey(config.getHost())){
            synchronized (poolMap){
                //当两个线程都来到synchronized这一行时，需要再代码块再次进行判断，否则会再次新建
                if (!poolMap.containsKey(config.getHost())){
                    poolMap.put(config.getHost(), new SshClientPool(config));
                }
            }
        }
        return poolMap.get(config.getHost());
    }

}

四、附加_多线程调试方法

    假设同一个host有两条线程同时进入创建对象连接池的方法，这里使用到了同步锁，此时需要调试是否只有一个线程进入到了同步代码块，可以使用线程调试方法。

（1）定义一个线程executor，并设置线程断点

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

public class SshExecutorThread implements Runnable {

    @Override
    public void run() {
        SshClientConfig sshClientConfig = new SshClientConfig();
        GenericObjectPoolConfig objectPoolConfig = new GenericObjectPoolConfig();

        objectPoolConfig.setMaxTotal(5);//最大连接数
        objectPoolConfig.setMaxIdle(3);//最大连接空闲数
        objectPoolConfig.setMinIdle(1);//最小连接空闲数
        objectPoolConfig.setMaxWaitMillis(-1);//连接超时时间，-1为默认不超时
        sshClientConfig.setObjectPoolConfig(objectPoolConfig);

        sshClientConfig.setHost("21.96.0.100");
        sshClientConfig.setPort(22);//远程连接默认端口
        sshClientConfig.setUser("hive");
        sshClientConfig.setRsaPath("D://key/id_rsa");
        sshClientConfig.setCommand("cat /user/hive/yarn.config");

        SshClientPool hostPool = SshClientHostPool.getHostPool(sshClientConfig);
        System.out.println("连接成功");

    }
}

（2）主线程同样也设置线程断点

（3）这样就可以切换线程进行调试了啊

终于写完了，希望大家多多支持，后面会分享更多精彩内容！！！！