记一次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; }}