聊聊jedis的borrow行为

熵桥冰川
• 阅读 883

本文主要研究一下jedis的borrow行为

borrowObject

org/apache/commons/pool2/impl/GenericObjectPool.java

    public T borrowObject(final Duration borrowMaxWaitDuration) throws Exception {
        assertOpen();

        final AbandonedConfig ac = this.abandonedConfig;
        if (ac != null && ac.getRemoveAbandonedOnBorrow() && (getNumIdle() < 2) &&
                (getNumActive() > getMaxTotal() - 3)) {
            removeAbandoned(ac);
        }

        PooledObject<T> p = null;

        // Get local copy of current config so it is consistent for entire
        // method execution
        final boolean blockWhenExhausted = getBlockWhenExhausted();

        boolean create;
        final long waitTimeMillis = System.currentTimeMillis();

        while (p == null) {
            create = false;
            p = idleObjects.pollFirst();
            if (p == null) {
                p = create();
                if (p != null) {
                    create = true;
                }
            }
            if (blockWhenExhausted) {
                if (p == null) {
                    if (borrowMaxWaitDuration.isNegative()) {
                        p = idleObjects.takeFirst();
                    } else {
                        p = idleObjects.pollFirst(borrowMaxWaitDuration);
                    }
                }
                if (p == null) {
                    throw new NoSuchElementException(appendStats(
                            "Timeout waiting for idle object, borrowMaxWaitDuration=" + borrowMaxWaitDuration));
                }
            } else if (p == null) {
                throw new NoSuchElementException(appendStats("Pool exhausted"));
            }
            if (!p.allocate()) {
                p = null;
            }

            if (p != null) {
                try {
                    factory.activateObject(p);
                } catch (final Exception e) {
                    try {
                        destroy(p, DestroyMode.NORMAL);
                    } catch (final Exception e1) {
                        // Ignore - activation failure is more important
                    }
                    p = null;
                    if (create) {
                        final NoSuchElementException nsee = new NoSuchElementException(
                                appendStats("Unable to activate object"));
                        nsee.initCause(e);
                        throw nsee;
                    }
                }
                if (p != null && getTestOnBorrow()) {
                    boolean validate = false;
                    Throwable validationThrowable = null;
                    try {
                        validate = factory.validateObject(p);
                    } catch (final Throwable t) {
                        PoolUtils.checkRethrow(t);
                        validationThrowable = t;
                    }
                    if (!validate) {
                        try {
                            destroy(p, DestroyMode.NORMAL);
                            destroyedByBorrowValidationCount.incrementAndGet();
                        } catch (final Exception e) {
                            // Ignore - validation failure is more important
                        }
                        p = null;
                        if (create) {
                            final NoSuchElementException nsee = new NoSuchElementException(
                                    appendStats("Unable to validate object"));
                            nsee.initCause(validationThrowable);
                            throw nsee;
                        }
                    }
                }
            }
        }

        updateStatsBorrow(p, Duration.ofMillis(System.currentTimeMillis() - waitTimeMillis));

        return p.getObject();
    }
  • borrowObject方法会开启一个while循环,条件是p为null,也就是要获取到p或者是内部自己跳出循环;idleObjects.pollFirst()从连接池获取,如果为null则执行create,之后是blockWhenExhausted的判断逻辑,如果create出来的为null,则阻塞等待takeFirst或者pollFirst(borrowMaxWaitDuration),如果还是null则抛出NoSuchElementException;如果blockWhenExhausted为false但是create为null则抛出Pool exhausted
  • 如果不是null,则再次确认下object的状态,如果变更状态(PooledObjectState.IDLE-->PooledObjectState.ALLOCATED)不成功则返回null;接着执行factory.activateObject(p)方法,如果出现异常则destory掉(jedis这里只是在db不一样的时候会重新select,默认可以理解为空操作),紧接着是testOnBorrow的逻辑
  • 这里就是如果idleObjects.pollFirst()为null会触发create,如果还是null则直接抛出NoSuchElementException异常,跳出循环;只有在不为null且allocate失败的时候会重置为null继续循环;另外如果是create出来的但是activate不成功也会抛出NoSuchElementException异常,跳出循环

create

    /**
     * Attempts to create a new wrapped pooled object.
     * <p>
     * If there are {@link #getMaxTotal()} objects already in circulation
     * or in process of being created, this method returns null.
     * </p>
     *
     * @return The new wrapped pooled object
     *
     * @throws Exception if the object factory's {@code makeObject} fails
     */
    private PooledObject<T> create() throws Exception {
        int localMaxTotal = getMaxTotal();
        // This simplifies the code later in this method
        if (localMaxTotal < 0) {
            localMaxTotal = Integer.MAX_VALUE;
        }

        final long localStartTimeMillis = System.currentTimeMillis();
        final long localMaxWaitTimeMillis = Math.max(getMaxWaitDuration().toMillis(), 0);

        // Flag that indicates if create should:
        // - TRUE:  call the factory to create an object
        // - FALSE: return null
        // - null:  loop and re-test the condition that determines whether to
        //          call the factory
        Boolean create = null;
        while (create == null) {
            synchronized (makeObjectCountLock) {
                final long newCreateCount = createCount.incrementAndGet();
                if (newCreateCount > localMaxTotal) {
                    // The pool is currently at capacity or in the process of
                    // making enough new objects to take it to capacity.
                    createCount.decrementAndGet();
                    if (makeObjectCount == 0) {
                        // There are no makeObject() calls in progress so the
                        // pool is at capacity. Do not attempt to create a new
                        // object. Return and wait for an object to be returned
                        create = Boolean.FALSE;
                    } else {
                        // There are makeObject() calls in progress that might
                        // bring the pool to capacity. Those calls might also
                        // fail so wait until they complete and then re-test if
                        // the pool is at capacity or not.
                        makeObjectCountLock.wait(localMaxWaitTimeMillis);
                    }
                } else {
                    // The pool is not at capacity. Create a new object.
                    makeObjectCount++;
                    create = Boolean.TRUE;
                }
            }

            // Do not block more if maxWaitTimeMillis is set.
            if (create == null &&
                (localMaxWaitTimeMillis > 0 &&
                 System.currentTimeMillis() - localStartTimeMillis >= localMaxWaitTimeMillis)) {
                create = Boolean.FALSE;
            }
        }

        if (!create.booleanValue()) {
            return null;
        }

        final PooledObject<T> p;
        try {
            p = factory.makeObject();
            if (getTestOnCreate() && !factory.validateObject(p)) {
                createCount.decrementAndGet();
                return null;
            }
        } catch (final Throwable e) {
            createCount.decrementAndGet();
            throw e;
        } finally {
            synchronized (makeObjectCountLock) {
                makeObjectCount--;
                makeObjectCountLock.notifyAll();
            }
        }

        final AbandonedConfig ac = this.abandonedConfig;
        if (ac != null && ac.getLogAbandoned()) {
            p.setLogAbandoned(true);
            p.setRequireFullStackTrace(ac.getRequireFullStackTrace());
        }

        createdCount.incrementAndGet();
        allObjects.put(new IdentityWrapper<>(p.getObject()), p);
        return p;
    }
create方法不会判断createCount,如果超出则返回null,如果等待超出maxWait也会返回null;如果判断要创建则通过factory.makeObject(),另外针对testOnCreate且validateObject不通过的也返回null,如果是有异常则直接抛出

makeObject

redis/clients/jedis/JedisFactory.java

  @Override
  public PooledObject<Jedis> makeObject() throws Exception {
    Jedis jedis = null;
    try {
      jedis = new Jedis(jedisSocketFactory, clientConfig);
      jedis.connect();
      return new DefaultPooledObject<>(jedis);
    } catch (JedisException je) {
      if (jedis != null) {
        try {
          jedis.quit();
        } catch (RuntimeException e) {
          logger.warn("Error while QUIT", e);
        }
        try {
          jedis.close();
        } catch (RuntimeException e) {
          logger.warn("Error while close", e);
        }
      }
      throw je;
    }
  }
JedisFactory的makeObject会创建Jedis然后执行connect,如果有JedisException则抛出,这个也会直接跳出borrowObject的循环,直接给到调用方

activateObject

redis/clients/jedis/JedisFactory.java

  public void activateObject(PooledObject<Jedis> pooledJedis) throws Exception {
    final BinaryJedis jedis = pooledJedis.getObject();
    if (jedis.getDB() != clientConfig.getDatabase()) {
      jedis.select(clientConfig.getDatabase());
    }
  }
JedisFactory的activateObject就是判断db跟配置的是不是一样,不一样则重新select

testOnBorrow

                if (p != null && getTestOnBorrow()) {
                    boolean validate = false;
                    Throwable validationThrowable = null;
                    try {
                        validate = factory.validateObject(p);
                    } catch (final Throwable t) {
                        PoolUtils.checkRethrow(t);
                        validationThrowable = t;
                    }
                    if (!validate) {
                        try {
                            destroy(p, DestroyMode.NORMAL);
                            destroyedByBorrowValidationCount.incrementAndGet();
                        } catch (final Exception e) {
                            // Ignore - validation failure is more important
                        }
                        p = null;
                        if (create) {
                            final NoSuchElementException nsee = new NoSuchElementException(
                                    appendStats("Unable to validate object"));
                            nsee.initCause(validationThrowable);
                            throw nsee;
                        }
                    }
                }

    public static void checkRethrow(final Throwable t) {
        if (t instanceof ThreadDeath) {
            throw (ThreadDeath) t;
        }
        if (t instanceof VirtualMachineError) {
            throw (VirtualMachineError) t;
        }
        // All other instances of Throwable will be silently swallowed
    }
testOnBorrow的逻辑就是执行validateObject方法,如果是ThreadDeath或者VirtualMachineError才会重新抛出,否则吞掉,之后判断validate结果,如果不成功则执行destory方法,重新设置为null,但是如果这个是create出来的则抛出NoSuchElementException

小结

jedis的borrow行为是在while循环里头去获取的,一般是在allocate变更状态不成功(PooledObjectState.IDLE-->PooledObjectState.ALLOCATED)的时候会重新设置null,继续循环

  • idleObjects.pollFirst()为null会触发create,如果还是null则抛出NoSuchElementException(Pool exhausted)跳出循环;如果blockWhenExhausted为true,block之后获取到的还是null,也会抛出NoSuchElementException(Timeout waiting for idle object)跳出循环;如果触发create操作,且create抛出JedisException,这个也会直接跳出borrowObject的循环,直接给到调用方
  • borrow出来不会null的执行activateObject,jedis这里只是在db不一样的时候会重新select,默认可以理解为空操作
  • 最后是testOnBorrow的逻辑,如果有异常,则针对create出来的则抛出NoSuchElementException跳出循环,否则重置为null继续循环

    总结一下就是如果是create有异常(JedisException)则直接抛出,如果borrow不到(即使经过create)也会抛出NoSuchElementException(具体可能是Pool exhausted或者Timeout waiting for idle object),如果有testOnBorrow不通过且是create出来的,也会抛出NoSuchElementException(Unable to validate object)
点赞
收藏
评论区
推荐文章
Wesley13 Wesley13
3年前
java Redis Jedis存储Java对象
Redis入门–Jedis存储Java对象(Java序列化为byte数组方式)Redis入门–Jedis存储Java对象\(Java序列化为byte数组方式)原文地址:http://alanland.iteye.com/admin/blogs/1600685(https://www.oschina.net/action/GoT
Wesley13 Wesley13
3年前
java使用jeids实现redis2.6的list操作(2)
代码:            https://github.com/viakiba/redis importorg.junit.Test;importredis.clients.jedis.BinaryClient.LIST_POSITION;importredis.clients.jedis.Jedis
Wesley13 Wesley13
3年前
Redis分布式锁
packagecom.sqi;importredis.clients.jedis.Jedis;importredis.clients.jedis.JedisPool;importredis.clients.jedis.params.SetParams;importjava.util.\;importjava.util.concur
Wesley13 Wesley13
3年前
java使用jeids实现redis2.6的key管理(2)
代码:    https://github.com/viakiba/redis主要是:    MIGRATE,MOVE,OBJECT,PERSIST/PEXPIRE/PEXPIREAT,PTTL/TTLimportorg.junit.Test;importredis.clients.jedis.Jedis;
Stella981 Stella981
3年前
Redis Cluster集群主从方案
RedisCluster集群主从方案本文介绍一种通过Jedis和Cluster实现Redis集群(主从)的高可用方案,该方案需要使用Jedis2.8.0(推荐),Redis3.0及以上版本(强制).附:RedisCluster集群主从方案:http://www.cnblogs.com/soul
Stella981 Stella981
3年前
SpringBoot+Redis分布式锁:模拟抢单
本篇内容主要讲解的是redis分布式锁,这个在各大厂面试几乎都是必备的,下面结合模拟抢单的场景来使用她;本篇不涉及到的redis环境搭建,快速搭建个人测试环境,这里建议使用docker;本篇内容节点如下:jedis的nx生成锁如何删除锁模拟抢单动作(10w个人开抢)jedis的nx生成锁
Stella981 Stella981
3年前
Redis应用学习(三)——Jedis客户端的简单使用介绍
1\.Redis的Java客户端程序——Jedis  1.在Redis的安装目录中rediscli是一个Redis的Shell语言写的客户端,通过运行该客户端就可以实现一系列Redis操作,而如果要使用Java语言的Redis客户端,就必须要使用Jedis相关的包。首先导入Jedis相关的包
Wesley13 Wesley13
3年前
J2Cache 中使用 Lettuce 替代 Jedis 管理 Redis 连接
一直以来J2Cache都是使用Jedis连接Redis服务的。Jedis是一个很老牌的Redis的Java开发包,使用很稳定,作者维护很勤勉,社区上能搜到的文章也非常非常多。算是使用范围最广的Redis开发包。但是Jedis比较推出时间比较早,整个设计思路比较传统,例如不支持异步操作,接口设计比较繁琐老套(相比其他开发包而已),
Easter79 Easter79
3年前
SpringBoot+Redis分布式锁:模拟抢单
本篇内容主要讲解的是redis分布式锁,这个在各大厂面试几乎都是必备的,下面结合模拟抢单的场景来使用她;本篇不涉及到的redis环境搭建,快速搭建个人测试环境,这里建议使用docker;本篇内容节点如下:jedis的nx生成锁如何删除锁模拟抢单动作(10w个人开抢)jedis的nx生成锁
Stella981 Stella981
3年前
Redis的Java客户端Jedis的八种调用方式(事务、管道、分布式)介绍
在这里对jedis关于事务、管道和分布式的调用方式做一个简单的介绍和对比:一、普通同步方式最简单和基础的调用方式,@Testpublic void test1Normal() {    Jedis jedis  new Jedis("localhost");    long start  Syst
Stella981 Stella981
3年前
Redis的Util工具类,超好用!
packagecom.lgdz.zhenshiUtil;importredis.clients.jedis.Jedis;importredis.clients.jedis.JedisPool;importredis.clients.jedis.JedisPoolConfig;importjava.