Redis连接池

一，Redis常见JAVA客户端

jedis

概况：Jedis是redis的Java客户端，其API提供了比较全面的Redis命令的支持。Jedis中的方法调用是比较底层的暴露的Redis的API，也即Jedis中的Java方法基本和Redis的API保持着一致；
优点：底层api比较全面
缺点：Jedis实例非线程安全，需要使用池化技术来使用jedis。
使用阻塞io，方法调用啥同步的，性能较差。

letture

概况：底层基于NETTY实现，是一种可扩展的线程安全的 Redis 客户端，支持异步模式，支持高级的 Redis 特性，比如哨兵，集群，管道，自动重新连接和Redis数据模型（springboot2默认使用letture）
优点：支持异步通信
实例线程安全

redission

概述：Redisson 是一个具有 In-Memory Data Grid 特性的 Redis Java 客户端。它提供了更方便和最简单的方式来使用 Redis。Redisson 对象提供了关注点分离，让您可以专注于数据建模和应用程序逻辑。
优点：对redis的高级封装，提供了很多分布式解决方案（分布式锁等）

二、Redis池化

1. 为什么要池化

1.1 jedis直连

在这里插入图片描述

1.2 jedis池化

在这里插入图片描述

1.3池化优势
  • 避免频繁的连接创建、释放的性能开销
  • 通过复用方式提高响应速度
  • 资源的分配、通过连接次更好的管理资源

2. 什么是池化

池化技术 (Pool) 是一种很常见的编程技巧，在请求量大时能明显优化应用性能，降低系统频繁建连的资源开销。我们日常工作中常见的有数据库连接池、线程池、对象池等，它们的特点都是将 “昂贵的”、“费时的” 的资源维护在一个特定的 “池子” 中，规定其最小连接数、最大连接数、阻塞队列等配置，方便进行统一管理和复用，通常还会附带一些探活机制、强制回收、监控一类的配套功能。

3. 池化框架commons-pool2解析

Apache commons-pool2类库是对象池技术的一种具体实现，它的出现是为了解决频繁的创建和销毁对象带来的性能损耗问题，原理就是建立一个对象池，池中预先生成了一些对象，需要对象的时候进行租借，用完后进行归还，对象不够时灵活的自动创建，对象池满后提供参数控制是否阻塞还是非阻塞响应租借.
熟悉了Apache commons-pool2对于了解数据库连接池DBCP和了解redis客户端jedis的连接池都有很大帮助，因为jedis的连接池就是基于Apache commons-pool2实现，而DBCP是基于Jakarta commons-pool实现。

3.1 commonspool2核心类
  • PooledObjectFactory：对象工厂，用户自己实现。负责对象等创建、初始化、销毁、验证等工作。
  • PooledObject：池化对象，是一个包装类，包装了池化的对象，并添加了一些附加信息如状态、创建时间、激活时间等。commons-pool2提供了DefaultPooledObject和 PoolSoftedObject 2种实现。其中PoolSoftedObject继承自DefaultPooledObject，不同点是使用SoftReference实现了对象的软引用。获取对象的时候使用也是通过SoftReference进行获取。
  • ObjectPool：对象池，负责对象生命周期对管理，并提供了对象池中活跃对象和空闲对象统计对功能。
3.2 对象池
3.2.1 对象池接口
public interface ObjectPool<T> extends Closeable {

	/**
	* 获取对象
	*/
    T borrowObject() throws Exception, NoSuchElementException, IllegalStateException;
	
	/**
	* 归还对象
	*/
    void returnObject(T var1) throws Exception;

	/**
	* 使对象失效，销毁对象
	*/
    void invalidateObject(T var1) throws Exception;

	/**
	* 添加对象
	*/
    void addObject() throws Exception, IllegalStateException, UnsupportedOperationException;

    int getNumIdle();

    int getNumActive();

    void clear() throws Exception, UnsupportedOperationException;

    void close();
}
3.2.2 类图

在这里插入图片描述

3.2.3 对象工厂
public interface PooledObjectFactory<T> {
	// 创建对象
    PooledObject<T> makeObject() throws Exception;
	//销毁对象
    void destroyObject(PooledObject<T> var1) throws Exception;
	//校验对象是否有效
    boolean validateObject(PooledObject<T> var1);
	//激活对象或启动对象的某些操作
    void activateObject(PooledObject<T> var1) throws Exception;
	//钝化对象，可以对对象做一些清理工作，比如清理过期数据
    void passivateObject(PooledObject<T> var1) throws Exception;
}
3.3 GenericObjectPool 详解

基于commons-pool2:2.6.2

3.3.1 类图

在这里插入图片描述

3.3.2 对象创建

Commons-pool2使用工厂模式实现对象池与对象的生成实现细节解耦，每一个对象池都有对象工厂作为成员变量。业务方在使用过程中需要自己去实现对象工厂。

public GenericObjectPool(PooledObjectFactory<T> factory) {
        this(factory, new GenericObjectPoolConfig());
    }

    public GenericObjectPool(PooledObjectFactory<T> factory, GenericObjectPoolConfig<T> 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.makeObjectCount = 0L;
        this.makeObjectCountLock = new Object();
        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);
        }
    }

    public GenericObjectPool(PooledObjectFactory<T> factory, GenericObjectPoolConfig<T> config, AbandonedConfig abandonedConfig) {
        this(factory, config);
        this.setAbandonedConfig(abandonedConfig);
    }

对象池的初始化要依赖对象工厂和连接池配置类，基于对象工厂来生成对象。我们接着看addObject方法。

public void addObject() throws Exception {
        this.assertOpen();
        if (this.factory == null) {
            throw new IllegalStateException("Cannot add objects without a factory.");
        } else {
        	//创建对象
            PooledObject<T> p = this.create();
            this.addIdleObject(p);
        }
 }

			//this.create();代码段
			PooledObject p;
            try {
                var20 = true;
                //调用对象工厂创建对象
                p = this.factory.makeObject();
                var20 = false;
            } catch (Throwable var23) {
                this.createCount.decrementAndGet();
                throw var23;
            }

所以对象池对象的创建依赖对象工厂

3.3.3 核心参数
	private volatile String factoryType;
    private volatile int maxIdle;
    private volatile int minIdle;
    //对象工厂
    private final PooledObjectFactory<T> factory;
    // 使用map存储所有对象，便于归还对象、销毁对象时找到PooledObject
    private final Map<IdentityWrapper<T>, PooledObject<T>> allObjects;
    private final AtomicLong createCount;
    private long makeObjectCount;
    private final Object makeObjectCountLock;
    //双端队列，空闲对象存储器
    private final LinkedBlockingDeque<PooledObject<T>> idleObjects;
    private static final String ONAME_BASE = "org.apache.commons.pool2:type=GenericObjectPool,name=";
    private volatile AbandonedConfig abandonedConfig;

如上核心参数中有个LinkedBlockingDeque双端阻塞队列，连接池使用这个队列来存储池对象。LinkedBlockingDeque是线程安全的阻塞队列，提供了FIFO、FILO策略。对象的出入队列都支持阻塞。

3.3.4 池化对象状态机

commons-pool2定义了如下状态枚举类。

public enum PooledObjectState {
	//空闲存活
    IDLE,
    //已分配
    ALLOCATED,
    //回收
    EVICTION,
    //
    EVICTION_RETURN_TO_HEAD,
    VALIDATION,//生效、相关状态暂未使用
    VALIDATION_PREALLOCATED,
    VALIDATION_RETURN_TO_HEAD,
    //失效
    INVALID,
    //被抛弃
    ABANDONED,
    //归还中
    RETURNING;

    private PooledObjectState() {
    }
}

PooledObject 定义了状态机。蓝色表示状态，白色表示方法。
在这里插入图片描述
这块我没有研究很深，感觉这块可以研究一下

3.3.5 对象池browObject
public T borrowObject(long borrowMaxWaitMillis) throws Exception {
        this.assertOpen();
        AbandonedConfig ac = this.abandonedConfig;
        // 判断是否调用removeAbandoned方法
        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;
                        //尝试从双端队列中获取对象，此方法非阻塞
                        p = (PooledObject)this.idleObjects.pollFirst();
                        if (p == null) {
                            p = this.create();
                            if (p != null) {
                                create = true;
                            }
                        }

                        if (blockWhenExhausted) {
                            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");
                            }
                        } else if (p == null) {
                            throw new NoSuchElementException("Pool exhausted");
                        }
						// 变更对象状态
                        if (!p.allocate()) {
                            p = null;
                        }
                    } while(p == null);

                    try {
                    	//调用工厂的activateObject方法，初始化激活对象
                        this.factory.activateObject(p);
                    } catch (Exception var13) {
                        try {
                        	//发生异常，销毁对象
                            this.destroy(p);
                        } catch (Exception var12) {
                        }

                        p = null;
                        if (create) {
                            NoSuchElementException nsee = new NoSuchElementException("Unable to activate object");
                            nsee.initCause(var13);
                            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 var15) {
                PoolUtils.checkRethrow(var15);
                validationThrowable = var15;
            }
			//对象不可用
            if (!validate) {
                try {
                	//销毁对象
                    this.destroy(p);
                    this.destroyedByBorrowValidationCount.incrementAndGet();
                } catch (Exception var14) {
                }

                p = null;
                if (create) {
                    NoSuchElementException nsee = new NoSuchElementException("Unable to validate object");
                    nsee.initCause(validationThrowable);
                    throw nsee;
                }
            }
        }
    }
3.3.6对象池returnObject
public void returnObject(T obj) {
        PooledObject<T> p = (PooledObject)this.allObjects.get(new IdentityWrapper(obj));
        if (p == null) {
            if (!this.isAbandonedConfig()) {
                throw new IllegalStateException("Returned object not currently part of this pool");
            }
        } else {
        	//标记状态归还中
            this.markReturningState(p);
            long activeTime = p.getActiveTimeMillis();
            //判断是否进行有效性验证
            if (this.getTestOnReturn() && !this.factory.validateObject(p)) {
                try {
                	//检查失败，销毁对象
                    this.destroy(p);
                } catch (Exception var10) {
                    this.swallowException(var10);
                }

                try {
                	//确保对象池中有idea状态对象可用，若没有则create
                    this.ensureIdle(1, false);
                } catch (Exception var9) {
                    this.swallowException(var9);
                }

                this.updateStatsReturn(activeTime);
            } else {
                try {
                	//返还对象后钝化对象
                    this.factory.passivateObject(p);
                } catch (Exception var12) {
                    this.swallowException(var12);

                    try {
                    	//异常销毁对象
                        this.destroy(p);
                    } catch (Exception var8) {
                        this.swallowException(var8);
                    }

                    try {
                        this.ensureIdle(1, false);
                    } catch (Exception var7) {
                        this.swallowException(var7);
                    }

                    this.updateStatsReturn(activeTime);
                    return;
                }
				//修改对象状态未idle
                if (!p.deallocate()) {
                    throw new IllegalStateException("Object has already been returned to this pool or is invalid");
                } else {
                    int maxIdleSave = this.getMaxIdle();
                    //判断对象是否超过最大空闲数
                    if (this.isClosed() || maxIdleSave > -1 && maxIdleSave <= this.idleObjects.size()) {
                        try {
                        	//超过销毁
                            this.destroy(p);
                        } catch (Exception var11) {
                            this.swallowException(var11);
                        }
                    } else {
                    	//没有超过，归还到队列里面去
                        if (this.getLifo()) {
                            this.idleObjects.addFirst(p);
                        } else {
                            this.idleObjects.addLast(p);
                        }

                        if (this.isClosed()) {
                            this.clear();
                        }
                    }

                    this.updateStatsReturn(activeTime);
                }
            }
        }
    }

确保对象池有活跃对象，如果没有则重新创建

private void ensureIdle(int idleCount, boolean always) throws Exception {
        if (idleCount >= 1 && !this.isClosed() && (always || this.idleObjects.hasTakeWaiters())) {
            while(this.idleObjects.size() < idleCount) {
                PooledObject<T> p = this.create();
                if (p == null) {
                    break;
                }

                if (this.getLifo()) {
                    this.idleObjects.addFirst(p);
                } else {
                    this.idleObjects.addLast(p);
                }
            }

            if (this.isClosed()) {
                this.clear();
            }

        }
    }
3.6.7 对象池保护机制
  1. 基于阈值的检测
// 判断是否调用removeAbandoned方法，启动对象泄漏清理
//当前对象池中少于两个idle对象或者active对象数大于（最大对象数-3）
        if (ac != null && ac.getRemoveAbandonedOnBorrow() && this.getNumIdle() < 2 && this.getNumActive() > this.getMaxTotal() - 3) {
            this.removeAbandoned(ac);
        }

在对象池获取对象的时候，会检测当前对象池活跃对象和空闲对象的占比。当空闲独享非常少、活跃非常多的时候。会触发空闲对象的回收。

  1. 异步调度线程检测
    在BaseGenericObjectPool抽象类中，有下面这段代码
public final void setTimeBetweenEvictionRunsMillis(long timeBetweenEvictionRunsMillis) {
        this.timeBetweenEvictionRunsMillis = timeBetweenEvictionRunsMillis;
        this.startEvictor(timeBetweenEvictionRunsMillis);
    }

在设置了TimeBetweenEvictionRunsMillis参数后会开启定时清理任务

final void startEvictor(long delay) {
        synchronized(this.evictionLock) {
            EvictionTimer.cancel(this.evictor, this.evictorShutdownTimeoutMillis, TimeUnit.MILLISECONDS);
            this.evictor = null;
            this.evictionIterator = null;
            //delay小于零，则不会开启定时任务
            if (delay > 0L) {
                this.evictor = new BaseGenericObjectPool.Evictor();
                EvictionTimer.schedule(this.evictor, delay, delay);
            }

        }
    }

接下来看一下evictor的核心逻辑

class Evictor implements Runnable {
        private ScheduledFuture<?> scheduledFuture;

        Evictor() {
        }

        public void run() {
            ClassLoader savedClassLoader = Thread.currentThread().getContextClassLoader();

            try {
                if (BaseGenericObjectPool.this.factoryClassLoader != null) {
                    ClassLoader cl = (ClassLoader)BaseGenericObjectPool.this.factoryClassLoader.get();
                    if (cl == null) {
                        this.cancel();
                        return;
                    }

                    Thread.currentThread().setContextClassLoader(cl);
                }

                try {
                	//回收
                    BaseGenericObjectPool.this.evict();
                } catch (Exception var9) {
                    BaseGenericObjectPool.this.swallowException(var9);
                } catch (OutOfMemoryError var10) {
                    var10.printStackTrace(System.err);
                }

                try {
                    BaseGenericObjectPool.this.ensureMinIdle();
                } catch (Exception var8) {
                    BaseGenericObjectPool.this.swallowException(var8);
                }
            } finally {
                Thread.currentThread().setContextClassLoader(savedClassLoader);
            }

        }

        void setScheduledFuture(ScheduledFuture<?> scheduledFuture) {
            this.scheduledFuture = scheduledFuture;
        }

        void cancel() {
            this.scheduledFuture.cancel(false);
        }
    }

public void evict() throws Exception {
        this.assertOpen();
        if (this.idleObjects.size() > 0) {
            PooledObject<T> underTest = null;
            EvictionPolicy<T> evictionPolicy = this.getEvictionPolicy();
            synchronized(this.evictionLock) {
                EvictionConfig evictionConfig = new EvictionConfig(this.getMinEvictableIdleTimeMillis(), this.getSoftMinEvictableIdleTimeMillis(), this.getMinIdle());
                boolean testWhileIdle = this.getTestWhileIdle();
                int i = 0;
                //获取每次检测样本数
                int m = this.getNumTests();

                while(true) {
                    if (i >= m) {
                        break;
                    }

                    if (this.evictionIterator == null || !this.evictionIterator.hasNext()) {
                        this.evictionIterator = new EvictionIterator(this, this.idleObjects);
                    }

                    if (!this.evictionIterator.hasNext()) {
                        return;
                    }

                    label81: {
                        try {
                            underTest = this.evictionIterator.next();
                        } catch (NoSuchElementException var15) {
                            --i;
                            this.evictionIterator = null;
                            break label81;
                        }

                        if (!underTest.startEvictionTest()) {
                            --i;
                        } else {
                            boolean evict;
                            try {
                                evict = evictionPolicy.evict(evictionConfig, underTest, this.idleObjects.size());
                            } catch (Throwable var14) {
                                PoolUtils.checkRethrow(var14);
                                this.swallowException(new Exception(var14));
                                evict = false;
                            }

                            if (evict) {
                            	//如果可以被回收，则直接调用destroy方法回收
                                this.destroy(underTest);
                                this.destroyedByEvictorCount.incrementAndGet();
                            } else {
                            	//进行空闲验证
                                if (testWhileIdle) {
                                    boolean active = false;

                                    try {
                                    	//尝试激活对象状态，激活失败则说明空闲对象已经失连
                                        this.factory.activateObject(underTest);
                                        active = true;
                                    } catch (Exception var13) {
                                    	//直接销毁
                                        this.destroy(underTest);
                                        this.destroyedByEvictorCount.incrementAndGet();
                                    }

                                    if (active) {
                                    //较验对象有效性
                                        if (!this.factory.validateObject(underTest)) {
                                        //对象不可用，销毁对象
                                            this.destroy(underTest);
                                            this.destroyedByEvictorCount.incrementAndGet();
                                        } else {
                                            try {
                                            //因为校验激活了空闲对象，分配了额外资源。调用passivateObject
                                                this.factory.passivateObject(underTest);
                                            } catch (Exception var12) {
                                                this.destroy(underTest);
                                                this.destroyedByEvictorCount.incrementAndGet();
                                            }
                                        }
                                    }
                                }

                                if (!underTest.endEvictionTest(this.idleObjects)) {
                                }
                            }
                        }
                    }

                    ++i;
                }
            }
        }

        AbandonedConfig ac = this.abandonedConfig;
        if (ac != null && ac.getRemoveAbandonedOnMaintenance()) {
            this.removeAbandoned(ac);
        }

    }

4. redis池化技术

redis连接池使用commons-pool2对象池方式实现。

三、REDIS连接池常见配置

1.常见配置（jedis）

在这里插入图片描述

2. springboot2.1 redis连接池核心参数

（1）application.yml配置文件

spring:
  redis:
    #    reids的连接ip
    host: zhouyangmin.redis.rds.aliyuncs.com
    # Redis默认情况下有16个分片，这里配置具体使用的分片，默认是0
    #database: 0
    # 连接超时时间（毫秒)
    timeout: 1000ms
    password: Zym06245719
    #  springboot2.0后默认使用lettuce连接redis,底层使用的是netty框架做支撑
    lettuce:
      pool:
        # 连接池中的最小空闲连接 默认 0
        min-idle: 2
        # 连接池最大阻塞等待时间（使用负值表示没有限制） 默认 -1
        max-wait: -1
        # 连接池最大连接数（使用负值表示没有限制） 默认 8
        max-active: 10
        # 连接池中的最大空闲连接 默认 8
        max-idle: 8
        # 进行空闲检测周期
        time-between-eviction-runs: 60000

看源码：RedisProperties.class、RedisAutoConfiguration

补充

redis连接数相关命令：
查看连接数：info clients
查看连接客户端：client list
查看最大连接数：config get maxclients

参考文档

https://www.bbsmax.com/A/6pdDNv9RJw/
https://blog.csdn.net/u_ascend/article/details/80594306
https://zhuanlan.zhihu.com/p/84481313

# redis # java
Logo
华为开发者空间

华为开发者空间，是为全球开发者打造的专属开发空间，汇聚了华为优质开发资源及工具，致力于让每一位开发者拥有一台云主机，基于华为根生态开发、创新。

更多推荐

cover

华为云数据仓库专业级开发者认证重磅发布

avatar 华为开发者空间
cover

一文带你搞懂GaussDB数据库性能调优

avatar 华为开发者空间
cover

GaussDB全密态数据库等值查询

avatar 华为开发者空间