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();
            // 沒有空閑對象則創(chuàng)建
            if (p == null) {
                p = create();
                if (p != null) {
                    create = true;
                }
            }
            if (p == null) {
                // borrowMaxWaitMillis默認值為-1
                if (borrowMaxWaitMillis < 0) {
                    // 線程?？煺绽锼械膁ubbo線程都卡在這里，這是個阻塞方法，如果隊列里沒有新的連接會一直等待下去
                    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 {
        // 創(chuàng)建redis連接，如果發(fā)生超時會拋出異常
        // 默認的connectionTimeout和soTimeout都是2秒
        p = factory.makeObject();
    } catch (Exception e) {
        createCount.decrementAndGet();
        // 這里會把異常繼續(xù)往上拋出
        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命令的執(zhí)行順序
    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;
    }
}