JedisPool源码解析以及Jedis从缓冲池获取-释放资源

本是古典何须时尚 2023-10-15 13:03 51阅读 0赞

## JedisPool源码解析 ##

**目录**

JedisPool源码解析

一、JedisPoolConfig

二、JedisPool

Jedis从缓存池中获取资源-销毁资源

一、从缓存中获取redis实例

二、Jedis与redis通信协议格式

三、释放缓存池中redis实例资源

--------------------

RedisClusterUtil工具类会用到，与下列相关联

package com.pcitc.dmb.common;
    
    import com.pcitc.dmb.common.util.AppPropertiesUtils;
    import com.pcitc.dmb.common.util.StringConstantsUtil;
    import com.pcitc.dmb.exception.BusinessException;
    import org.apache.commons.lang3.StringUtils;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import redis.clients.jedis.*;
    
    import java.io.IOException;
    import java.util.HashSet;
    import java.util.List;
    import java.util.Set;
    
    /**
     * @Description: redis-Cluster工具类$
     * @Date: 2023-08-11$
     * @Param: $
     * @return: $
     */
    public class RedisClusterUtil {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(RedisClusterUtil.class);
        private static JedisCluster jedisCluster;
        private static JedisPool jedisPool;
        // 连接实例的最大连接数
        private static int MAX_ACTIVE = 10;
        // 控制一个pool最多有多少个状态为idle(空闲的)的jedis实例，默认值也是8。
        private static int MAX_IDLE =10;
        private static int MIN_IDLE = 2;
        private static int MAX_WAIT_MILLS = 30*1000;
        private static int TIME_BETWEEN_EVICTION_RUNS_MILLSIS = 10*1000;
        private static int MIN_EVICTABLE_IDLE_TIME_MILLS = 30*1000;
    
        /**
         * @Description: 初始化Redis连接池
         * @Date: 2023/8/28
         * @return:
         * @param:
         **/
        static {
            try{
                AppPropertiesUtils.initAppProperties("application.properties");
                //Redis集群节点信息
                Set<HostAndPort> nodes = new HashSet<>();
    
                /**
                 * @Description: 此处适配一个服务的六个端口的redis集群，本次在85服务器上部署了六个节点
                 * @Date: 2023/8/11
                 * @return:
                 * @param:
                 **/
                String node = AppPropertiesUtils.getProperty("redis.node1");
    
                if (StringUtils.isNotBlank(node)){
                    nodes.add(new HostAndPort(node,Integer.parseInt(AppPropertiesUtils.getProperty("redis.port1"))));
                    nodes.add(new HostAndPort(node,Integer.parseInt(AppPropertiesUtils.getProperty("redis.port2"))));
                    nodes.add(new HostAndPort(node,Integer.parseInt(AppPropertiesUtils.getProperty("redis.port3"))));
                    nodes.add(new HostAndPort(node,Integer.parseInt(AppPropertiesUtils.getProperty("redis.port4"))));
                    nodes.add(new HostAndPort(node,Integer.parseInt(AppPropertiesUtils.getProperty("redis.port5"))));
                    nodes.add(new HostAndPort(node,Integer.parseInt(AppPropertiesUtils.getProperty("redis.port6"))));
                }
    
                //Redis连接池配置
                JedisPoolConfig poolConfig = new JedisPoolConfig();
                poolConfig.setMaxTotal(MAX_ACTIVE);//最大连接数 设置最大实例总数
                poolConfig.setMaxIdle(MAX_IDLE);//最大空闲连接数
                poolConfig.setMinIdle(MIN_IDLE);//最小空闲连接数
                //表示当borrow（引入）一个jedis实例时，最大的等待时间，如果超过等待时间，则直接抛出JedisConnectionException；
                poolConfig.setMaxWaitMillis(MAX_WAIT_MILLS);
    
                //向资源池借用连接时是否做连接有效性检测（ping）。检测到的无效连接将会被移除。
                poolConfig.setTestOnBorrow(true);
                // 向资源池归还连接时是否做有效性监测（ping）。监测到无效会被移除
                poolConfig.setTestOnReturn(true);
                //是否开启空闲资源监测,这一项只有在timeBetweenEvictionRunsMillis大于0时才有意义；
                poolConfig.setTestWhileIdle(true);
                //资源池中资源最小空闲时间，达到此值后空闲资源将会被移除,这一项只有在timeBetweenEvictionRunsMillis大于0时才有意义；
                poolConfig.setMinEvictableIdleTimeMillis(MIN_EVICTABLE_IDLE_TIME_MILLS);
                //表示idle（空闲） object evitor两次扫描之间要sleep的毫秒数
                //空闲资源的监测周期
                poolConfig.setTimeBetweenEvictionRunsMillis(TIME_BETWEEN_EVICTION_RUNS_MILLSIS);
                //当资源池用尽后，调用者是否要等待，只有当值为true时，maxWaitMillis才会生效。
                poolConfig.setBlockWhenExhausted(true);
                //是否开启JMX监控
                poolConfig.setJmxEnabled(true);
    
                //创建JedisCluster
                jedisCluster = new JedisCluster(nodes,poolConfig);
                jedisPool = new JedisPool(poolConfig, node, 6380);
    
            } catch (Exception e){
                jedisCluster = null;
                LOGGER.error("redis集群连接异常，异常信息:{}", e.getMessage());
            }
        }
    
        /**
         * @Description: 获取JedisCluster实例
         * @Date: 2023/8/11 
         * @return: redis.clients.jedis.JedisCluster 
         * @param: []       
         **/
        public synchronized static JedisCluster getJedisCluster() {
            try{
                return jedisCluster;
            }catch (Exception e) {
                LOGGER.error("redis集群连接异常，异常信息:{}", e.getMessage());
                return null;
            }
        }
    
        /***
         *
         * 释放资源
         */
        @SuppressWarnings("deprecation")
        public static void releaseJedisCluster(JedisCluster jedisCluster) throws IOException {
            if (jedisCluster != null) {
                jedisCluster.close();
            }
        }
    
        public synchronized static Jedis getJedis() {
            try {
                if (jedisPool != null) {
                    Jedis resource = jedisPool.getResource();
                    LOGGER.info("redis--服务正在运行: " + resource.ping());
    
                    //当前活跃连接数
                    int numActive = jedisPool.getNumActive();
                    LOGGER.info(String.valueOf("当前活跃连接数: " + numActive));
                    //当前空闲连接数
                    int numIdle = jedisPool.getNumIdle();
                    LOGGER.info(String.valueOf("当前空闲连接数: " + numIdle));
                    return resource;
                } else {
                    return null;
                }
    
            } catch (Exception e) {
                LOGGER.error("redis连接异常，异常信息:{}", e.getMessage());
                return null;
            }
    
        }
    
        /**
         * @Description: 释放资源
         * @Date: 2023/8/28 
         * @return: void 
         * @param: [jedis]       
         **/
        @SuppressWarnings("deprecation")
        public static void releaseJedis(Jedis jedis){
            if(jedis != null){
                jedis.close();
            }
        }
    
        public static void main(String[] args) throws BusinessException, IOException {
    
            for (int i=1;i<13;i++){
                JedisCluster jedisCluster = RedisClusterUtil.getJedisCluster();
                Jedis jedis = RedisClusterUtil.getJedis();
                if (jedisCluster == null){
                    throw new BusinessException("","","redis集群连接异常");
                }
    
                if (jedis == null){
                    throw new BusinessException("","","redis连接异常");
                }
    
                // 使用JedisCluster对象进行Redis操作
                List<String> pidAssociation = jedisCluster.lrange(StringConstantsUtil.REDIS_PID_ASSOCIATION, 0, -1);
    //            LOGGER.info(String.valueOf(pidAssociation));
    
                LOGGER.info(String.valueOf(i));
            }
    
            // 释放资源
    //        RedisClusterUtil.releaseJedisCluster(jedisCluster);
    //        RedisClusterUtil.releaseJedis(jedis);
        }
    }

![img][]

### 一、JedisPoolConfig ###

**JedisPoolConfig的继承结构**

这个BaseObjectPoolConfig中无非是一部分常量和私有的变量。

public abstract class BaseObjectPoolConfig implements Cloneable {
        public static final boolean DEFAULT_LIFO = true;
        public static final boolean DEFAULT_FAIRNESS = false;
        public static final long DEFAULT_MAX_WAIT_MILLIS = -1L;
        public static final long DEFAULT_MIN_EVICTABLE_IDLE_TIME_MILLIS = 1800000L;
        public static final long DEFAULT_SOFT_MIN_EVICTABLE_IDLE_TIME_MILLIS = -1L;
        public static final int DEFAULT_NUM_TESTS_PER_EVICTION_RUN = 3;
        public static final boolean DEFAULT_TEST_ON_CREATE = false;
        public static final boolean DEFAULT_TEST_ON_BORROW = false;
        public static final boolean DEFAULT_TEST_ON_RETURN = false;
        public static final boolean DEFAULT_TEST_WHILE_IDLE = false;
        public static final long DEFAULT_TIME_BETWEEN_EVICTION_RUNS_MILLIS = -1L;
        public static final boolean DEFAULT_BLOCK_WHEN_EXHAUSTED = true;
        public static final boolean DEFAULT_JMX_ENABLE = true;
        public static final String DEFAULT_JMX_NAME_PREFIX = "pool";
        public static final String DEFAULT_JMX_NAME_BASE = null;
        public static final String DEFAULT_EVICTION_POLICY_CLASS_NAME = "org.apache.commons.pool2.impl.DefaultEvictionPolicy";
        private boolean lifo = true;
        private boolean fairness = false;
        private long maxWaitMillis = -1L;
        private long minEvictableIdleTimeMillis = 1800000L;
        private long softMinEvictableIdleTimeMillis = 1800000L;
        private int numTestsPerEvictionRun = 3;
        private String evictionPolicyClassName = "org.apache.commons.pool2.impl.DefaultEvictionPolicy";
        private boolean testOnCreate = false;
        private boolean testOnBorrow = false;
        private boolean testOnReturn = false;
        private boolean testWhileIdle = false;
        private long timeBetweenEvictionRunsMillis = -1L;
        private boolean blockWhenExhausted = true;
        private boolean jmxEnabled = true;
        private String jmxNamePrefix = "pool";
        private String jmxNameBase = "pool";

然后它的子类补充了部分的常量和变量。

public class GenericObjectPoolConfig extends BaseObjectPoolConfig {
        public static final int DEFAULT_MAX_TOTAL = 8;
        public static final int DEFAULT_MAX_IDLE = 8;
        public static final int DEFAULT_MIN_IDLE = 0;
        private int maxTotal = 8;
        private int maxIdle = 8;
        private int minIdle = 0;

然后实际初始化的是其构造方法。都是写死的。

public JedisPoolConfig() {
            this.setTestWhileIdle(true);
            this.setMinEvictableIdleTimeMillis(60000L);
            this.setTimeBetweenEvictionRunsMillis(30000L);
            this.setNumTestsPerEvictionRun(-1);
        }

例子

JedisPoolConfig config = new JedisPoolConfig();
            config.setTestOnBorrow(true);
            pool = new JedisPool(config, "", 2181, 2000, "foobared");
            jedis = pool.getResource();
            jedis.set("foo", "bar");
            jedis.close();

Config部分创建了一个配置项实例。

设置部分参数可以。

然后这个地方创建了一个JedisPool的实例，这个实例，

public JedisPool(GenericObjectPoolConfig poolConfig, String host, int port, int timeout, String password, int database) {
            this(poolConfig, host, port, timeout, password, database, (String)null);
        }
     
        public JedisPool(GenericObjectPoolConfig poolConfig, String host, int port, int timeout, String password, int database, String clientName) {
            super(poolConfig, new JedisFactory(host, port, timeout, password, database, clientName));
        }

### 二、JedisPool ###

**JedisPool->pool(初始化中调用GenericObjectPool)->GenericObjectPool**

JedisPool调用了父类的方法

![img][img 1]

集成体系，JedisPool继承自Pool这个类，

public class JedisPool extends Pool<Jedis> {}

public Pool(GenericObjectPoolConfig poolConfig, PooledObjectFactory<T> factory) {
        this.initPool(poolConfig, factory);
    }
     
    public void initPool(GenericObjectPoolConfig poolConfig, PooledObjectFactory<T> factory) {
        if (this.internalPool != null) {
            try {
                this.closeInternalPool();
            } catch (Exception var4) {
                ;
            }
        }
        this.internalPool = new GenericObjectPool(factory, poolConfig);
    }

调用Pool中的初始化pool方法。这个方法中调用了**GenericObjectPool**

public GenericObjectPool(PooledObjectFactory<T> factory, GenericObjectPoolConfig config) {
            super(config, "org.apache.commons.pool2:type=GenericObjectPool,name=", config.getJmxNamePrefix());
            this.factoryType = null;
            this.maxIdle = 8;
            this.minIdle = 0;
            this.allObjects = new ConcurrentHashMap();
            this.createCount = new AtomicLong(0L);
            this.abandonedConfig = null;
            if (factory == null) {
                this.jmxUnregister();
                throw new IllegalArgumentException("factory may not be null");
            } else {
                this.factory = factory;
                this.idleObjects = new LinkedBlockingDeque(config.getFairness());
                this.setConfig(config);
                this.startEvictor(this.getTimeBetweenEvictionRunsMillis());
            }
        }

初始化的GenericObjectPool

注意这里的重点是使用了LinkedBlockingDeque了，也就是双向链表实现的双端队列。空闲的连接放在这里面。

**JedisPool中的方法。getResource，部分获取空闲的连接；**

//JedisPool中
    public Jedis getResource() {
            Jedis jedis = (Jedis)super.getResource();
            jedis.setDataSource(this);
            return jedis;
        }
    
    
    // Pool中
    public T getResource() {
            try {
                return this.internalPool.borrowObject();
            } catch (NoSuchElementException var2) {
                throw new JedisException("Could not get a resource from the pool", var2);
            } catch (Exception var3) {
                throw new JedisConnectionException("Could not get a resource from the pool", var3);
            }
        }

调用GenericObjectPool中的方法borrowObject()

@Override
    public T borrowObject() throws Exception {
        return this.borrowObject(this.getMaxWaitMillis());
    }
    
    
    public T borrowObject() throws Exception {
        return this.borrowObject(this.getMaxWaitMillis());
    }
    
    public T borrowObject(long borrowMaxWaitMillis) throws Exception {
        this.assertOpen();
        AbandonedConfig ac = this.abandonedConfig;
        if (ac != null && ac.getRemoveAbandonedOnBorrow() && this.getNumIdle() < 2 && this.getNumActive() > this.getMaxTotal() - 3) {
            this.removeAbandoned(ac);
        }
    
        PooledObject<T> p = null;
        boolean blockWhenExhausted = this.getBlockWhenExhausted();
        long waitTime = System.currentTimeMillis();
    
        while(true) {
            boolean create;
            do {
                do {
                    do {
                        if (p != null) {
                            this.updateStatsBorrow(p, System.currentTimeMillis() - waitTime);
                            return p.getObject();
                        }
    
                        create = false;
                        if (blockWhenExhausted) {
                            p = (PooledObject)this.idleObjects.pollFirst();
                            if (p == null) {
                                p = this.create();
                                if (p != null) {
                                    create = true;
                                }
                            }
    
                            if (p == null) {
                                if (borrowMaxWaitMillis < 0L) {
                                    p = (PooledObject)this.idleObjects.takeFirst();
                                } else {
                                    p = (PooledObject)this.idleObjects.pollFirst(borrowMaxWaitMillis, TimeUnit.MILLISECONDS);
                                }
                            }
    
                            if (p == null) {
                                throw new NoSuchElementException("Timeout waiting for idle object");
                            }
    
                            if (!p.allocate()) {
                                p = null;
                            }
                        } else {
                            p = (PooledObject)this.idleObjects.pollFirst();
                            if (p == null) {
                                p = this.create();
                                if (p != null) {
                                    create = true;
                                }
                            }
    
                            if (p == null) {
                                throw new NoSuchElementException("Pool exhausted");
                            }
    
                            if (!p.allocate()) {
                                p = null;
                            }
                        }
                    } while(p == null);
    
                    try {
                        this.factory.activateObject(p);
                    } catch (Exception var15) {
                        try {
                            this.destroy(p);
                        } catch (Exception var14) {
                        }
    
                        p = null;
                        if (create) {
                            NoSuchElementException nsee = new NoSuchElementException("Unable to activate object");
                            nsee.initCause(var15);
                            throw nsee;
                        }
                    }
                } while(p == null);
            } while(!this.getTestOnBorrow() && (!create || !this.getTestOnCreate()));
    
            boolean validate = false;
            Throwable validationThrowable = null;
    
            try {
                validate = this.factory.validateObject(p);
            } catch (Throwable var13) {
                PoolUtils.checkRethrow(var13);
                validationThrowable = var13;
            }
    
            if (!validate) {
                try {
                    this.destroy(p);
                    this.destroyedByBorrowValidationCount.incrementAndGet();
                } catch (Exception var12) {
                }
    
                p = null;
                if (create) {
                    NoSuchElementException nsee = new NoSuchElementException("Unable to validate object");
                    nsee.initCause(validationThrowable);
                    throw nsee;
                }
            }
        }
    }

然后剩下的就是jedis本身的操作了。

总结：

最后理清思路，首先是创建一个JedisPool的实例，这个实例会调用父类Pool中的方法来创建Pool，然后这个Pool中有一个internalPool，使得其等于一个GenericJedisPool，然后这个pool实际上空闲的会存储在LinkedBlockingDeque中去

然后在getResource的时候，会调jedisPool的方法然后调用父类（Pool）的方法然后调用GenericPool的方法，然后就可以直接获取到空闲线程了。

## Jedis从缓存池中获取资源-销毁资源 ##

### 一、从缓存中获取redis实例 ###

通过JedisPool的getResource就可以从缓存池中取出一个redis实例对象，该方法是从Pool[类继承][Link 1]而来

@SuppressWarnings("unchecked")
        public T getResource() {
            try {
                return (T) internalPool.borrowObject();
            } catch (Exception e) {
                throw new JedisConnectionException(
                        "Could not get a resource from the pool", e);
            }
        }

在RedisClusterUtil工具类中会配置连接池（JedisPoolConfig）相关参数，其中就有testOnBorrow

在初始化JedisPool实例时，如果testOnBorrow为true的话，那么在borrowObject方法中会调用上文中提到的**JedisFactory**的**validateObject**方法来确报从borrowObject方法中获取到的实例对象必然是可用的。

public boolean validateObject(final Object obj) {
                if (obj instanceof Jedis) {
                    final Jedis jedis = (Jedis) obj;
                    try {
                        return jedis.isConnected() && jedis.ping().equals("PONG");
                    } catch (final Exception e) {
                        return false;
                    }
                } else {
                    return false;
                }
            }
        }

在validateObject方法中，Jedis会去看从缓存池中取出的redis实例是否和redis服务器保持连接，以及输入和输出流是否可用

public boolean isConnected() {
            return this.socket != null && this.socket.isBound() && !this.socket.isClosed() && this.socket.isConnected() && !this.socket.isInputShutdown() && !this.socket.isOutputShutdown();
        }

如果socket可用的话，那么再像redis服务器发送ping命令，测试与服务器的连接是否仍然生效

public String ping() {
            this.checkIsInMultiOrPipeline();
            this.client.ping();
            return this.client.getStatusCodeReply();
        }

方法中首先检查是否开启了事务，Jedis类并不支持事务，使用事务可用Transaction类，然后向redis服务器发送ping命令

### 二、Jedis与redis通信协议格式 ###

**Jedis向服务器发送命令最终是由Protocol类完成**

public static void sendCommand(RedisOutputStream os, Protocol.Command command, byte[]... args) {
        sendCommand(os, command.raw, args);
    }
    
    private static void sendCommand(RedisOutputStream os, byte[] command, byte[]... args) {
        try {
            os.write((byte)42);
            os.writeIntCrLf(args.length + 1);
            os.write((byte)36);
            os.writeIntCrLf(command.length);
            os.write(command);
            os.writeCrLf();
            byte[][] var3 = args;
            int var4 = args.length;
    
            for(int var5 = 0; var5 < var4; ++var5) {
                byte[] arg = var3[var5];
                os.write((byte)36);
                os.writeIntCrLf(arg.length);
                os.write(arg);
                os.writeCrLf();
            }
    
        } catch (IOException var7) {
            throw new JedisConnectionException(var7);
        }
    }

Jedis向服务器发送命令最终是由Protocol类完成，Jedis将创建时保留下来的输入流和要发送的命令以及参数传给Protocol的sendCommand方法，由Protocol类来向redis服务器发送通信内容。这个地方用到了命令模式这样的设计思想，Client类发送命令给Connection，Connection将命令传递给Protocol，由Protocol来决定命令具体怎么执行，这样Client和Protocol消除了耦合。

redis通信协议如下；

*<参数数量> CR LF
    $<参数 1 的字节数量> CR LF
    <参数 1 的数据> CR LF
    ...
    $<参数 N 的字节数量> CR LF
    <参数 N 的数据> CR LF```  
    注：命令本身也作为协议的其中一个参数来发送。
    2、回复
    redis回复格式如下
        状态回复（status reply）的第一个字节是 "+"
        错误回复（error reply）的第一个字节是 "-"
        整数回复（integer reply）的第一个字节是 ":"
        批量回复（bulk reply）的第一个字节是 "$"
        多条批量回复（multi bulk reply）的第一个字节是 "*"

private static Object process(RedisInputStream is) {
            byte b = is.readByte();
            if (b == 43) {
                return processStatusCodeReply(is);
            } else if (b == 36) {
                return processBulkReply(is);
            } else if (b == 42) {
                return processMultiBulkReply(is);
            } else if (b == 58) {
                return processInteger(is);
            } else if (b == 45) {
                processError(is);
                return null;
            } else {
                throw new JedisConnectionException("Unknown reply: " + (char)b);
            }
        }

通过Jedis创建时保留下来的输入流，来读取第一个字节，判断是那种类型的类型，然后进行相应的解析，以ping命令为例。

private static byte[] processStatusCodeReply(final RedisInputStream is) {
        return SafeEncoder.encode(is.readLine());
        }```

ping命令回复的类型属于状态回复。Jedis读取输入流中的第一行中除去第一个字节剩下的字节进行utf-8编码转换成字符串
    #

### 三、释放缓存池中redis实例资源 ###

/**
          * 释放jedis资源
          * @param jedis
          */
         public static void returnResource(final Jedis jedis) {
             if (jedis != null) {
                 jedisPool.returnResource(jedis);
             }
         }```

调用了基类Pool的returnResource方法。该方法会调用初始化JedisPool时传入的JedisFactory中的destroyObject方法来销毁资源。

public void destroyObject(PooledObject<Jedis> pooledJedis) throws Exception {
            BinaryJedis jedis = (BinaryJedis)pooledJedis.getObject();
            if (jedis.isConnected()) {
                try {
                    try {
                        jedis.quit();
                    } catch (Exception var4) {
                    }
    
                    jedis.disconnect();
                } catch (Exception var5) {
                }
            }
    
        }

改方法首先向redis服务器发送quit命令，来结束此会话。然后调用disconnect方法来关闭输入和输出流。

public void disconnect() {
            if (this.isConnected()) {
                try {
                    this.outputStream.flush();
                    this.socket.close();
                } catch (IOException var5) {
                    this.broken = true;
                    throw new JedisConnectionException(var5);
                } finally {
                    IOUtils.closeQuietly(this.socket);
                }
            }
    
        }

[img]: https://img-blog.csdn.net/20170727010657408?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvd2FuZ3lhbmcxMzU0/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center
[img 1]: https://img-blog.csdn.net/20170727225448660?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvd2FuZ3lhbmcxMzU0/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center
[Link 1]: https://so.csdn.net/so/search?q=%E7%B1%BB%E7%BB%A7%E6%89%BF&spm=1001.2101.3001.7020