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

JedisClusterPipline jedisClusterPipline = redisService.clusterPipelined();ListObject> 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 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 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 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(p.getObject()), p);    return p;}

JedisClusterPipline jedisClusterPipline = redisService.clusterPipelined();ListObject> 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 orderedClients = new LinkedList();    // redis连接缓存    private final Map 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{        List        List 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;    }}