记一次Redis Cluster Pipeline导致的死锁问题

JedisClusterPipline jedisClusterPipline = redisService.clusterPipelined();
List<Object> response;
try {
    for (String key : keys) {
        jedisClusterPipline.hmget(key, VALUE1, VALUE2);
    }
    // 获取结果
    response = jedisClusterPipline.syncAndReturnAll();
} finally {
    jedisClusterPipline.close();
}

public T borrowObject(long borrowMaxWaitMillis) throws Exception {
    ...
    PooledObject<T> p = null;
    // 获取blockWhenExhausted配置项，该配置默认值为true
    boolean blockWhenExhausted = getBlockWhenExhausted();
    boolean create;
    long waitTime = System.currentTimeMillis();
 
    while (p == null) {
        create = false;
        if (blockWhenExhausted) {
            // 从队列获取空闲的对象，该方法不会阻塞，没有空闲对象会返回null
            p = idleObjects.pollFirst();
            // 没有空闲对象则创建
            if (p == null) {
                p = create();
                if (p != null) {
                    create = true;
                }
            }
            if (p == null) {
                // borrowMaxWaitMillis默认值为-1
                if (borrowMaxWaitMillis < 0) {
                    // 线程栈快照里所有的dubbo线程都卡在这里，这是个阻塞方法，如果队列里没有新的连接会一直等待下去
                    p = idleObjects.takeFirst();    
                } else {
                    // 等待borrowMaxWaitMillis配置的时间还没有拿到连接的话就返回null
                    p = idleObjects.pollFirst(borrowMaxWaitMillis,
                            TimeUnit.MILLISECONDS);
                }
            }
            if (p == null) {
                throw new NoSuchElementException(
                        "Timeout waiting for idle object");
            }
            if (!p.allocate()) {
                p = null;
            }
        }
        ...
    }
    updateStatsBorrow(p, System.currentTimeMillis() - waitTime);
    return p.getObject();
}

private PooledObject<T> create() throws Exception {
    int localMaxTotal = getMaxTotal();
    long newCreateCount = createCount.incrementAndGet();
    if (localMaxTotal > -1 && newCreateCount > localMaxTotal ||
            newCreateCount > Integer.MAX_VALUE) {
        createCount.decrementAndGet();
        return null;
    }
    final PooledObject<T> p;
    try {
        // 创建redis连接，如果发生超时会抛出异常
        // 默认的connectionTimeout和soTimeout都是2秒
        p = factory.makeObject();
    } catch (Exception e) {
        createCount.decrementAndGet();
        // 这里会把异常继续往上抛出
        throw e;
    }
    AbandonedConfig ac = this.abandonedConfig;
    if (ac != null && ac.getLogAbandoned()) {
        p.setLogAbandoned(true);
    }
    createdCount.incrementAndGet();
    allObjects.put(new IdentityWrapper<T>(p.getObject()), p);
    return p;
}

JedisClusterPipline jedisClusterPipline = redisService.clusterPipelined();
List<Object> response;
try {
    for (String key : keys) {
        // 申请连接，内部会先调用JedisClusterPipeline.getClient(String key)方法获取连接
        jedisClusterPipline.hmget(key, VALUE1, VALUE2);
        // 获取到了连接，缓存到poolToJedisMap
    }
    // 获取结果
    response = jedisClusterPipline.syncAndReturnAll();
} finally {
    // 归还所有连接
    jedisClusterPipline.close();
}

public class JedisClusterPipline extends PipelineBase implements Closeable {
    private static final Logger log = LoggerFactory.getLogger(JedisClusterPipline.class);
    // 用于记录redis命令的执行顺序
    private final Queue<Client> orderedClients = new LinkedList<>();
    // redis连接缓存
    private final Map<JedisPool, Jedis> poolToJedisMap = new HashMap<>();
    private final JedisSlotBasedConnectionHandler connectionHandler;
    private final JedisClusterInfoCache clusterInfoCache;
    public JedisClusterPipline(JedisSlotBasedConnectionHandler connectionHandler, JedisClusterInfoCache clusterInfoCache) {
        this.connectionHandler = connectionHandler;
        this.clusterInfoCache = clusterInfoCache;
    }
    @Override
    protected Client getClient(String key) {
 
        return getClient(SafeEncoder.encode(key));
    }
    @Override
    protected Client getClient(byte[] key) {
        Client client;
        // 计算key所在的slot
        int slot = JedisClusterCRC16.getSlot(key);
        // 获取solt对应的连接池
        JedisPool pool = clusterInfoCache.getSlotPool(slot);
        // 从缓存获取连接
        Jedis borrowedJedis = poolToJedisMap.get(pool);
        // 缓存中没有连接则从连接池获取并缓存
        if (null == borrowedJedis) {
            borrowedJedis = pool.getResource();
            poolToJedisMap.put(pool, borrowedJedis);
        }
        client = borrowedJedis.getClient();
        orderedClients.add(client);
        return client;
    }
    @Override
    public void close() {
        for (Jedis jedis : poolToJedisMap.values()) {
            // 清除连接内的残留数据,防止连接归还时有数据漏读的现象
            try {
                jedis.getClient().getAll();
            } catch (Throwable throwable) {
                log.warn("关闭jedis时遍历异常，遍历的目的是：清除连接内的残留数据,防止连接归还时有数据漏读的现象");
            }
            try {
                jedis.close();
            } catch (Throwable throwable) {
                log.warn("关闭jedis异常");
            }
        }
        // 归还连接
        clean();
        orderedClients.clear();
        poolToJedisMap.clear();
    }
    /**
     * go through all the responses and generate the right response type (warning :
     * usually it is a waste of time).
     *
     * @return A list of all the responses in the order
     */
    public List<Object> syncAndReturnAll() {
        List<Object> formatted = new ArrayList<>();
        List<Throwable> throwableList = new ArrayList<>();
        for (Client client : orderedClients) {
            try {
                Response response = generateResponse(client.getOne());
                if(response == null){
                    continue;
                }
                formatted.add(response.get());
            } catch (Throwable e) {
                throwableList.add(e);
            }
        }
        slotCacheRefreshed(throwableList);
        return formatted;
    }
}