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

JedisClusterPipline jedisClusterPipline = redisService.clusterPipelined();
List

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 

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();
List

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