尝试获取锁对象, 不会阻塞

@throws InterruptedException
@throws KeeperException
347157
public boolean tryLock() throws KeeperException {
        // 可能初始化的时候就失败了
        if (exception != null) {
            throw exception;
        }

        if (isOwner()) {// 锁重入
            return true;
        }

        acquireLock(null);

        if (exception != null) {
            unlock();
            throw exception;
        }

        if (interrupt != null) {
            unlock();
            Thread.currentThread().interrupt();
        }

        if (other != null) {
            unlock();
            throw new NestableRuntimeException(other);
        }

        return isOwner();
    }
971345
@Override
    public T take(final WaitStrategy.Takeable<T> t) throws InterruptedException {
        T result;

        while ((result = t.take()) == null) {
            Thread.yield();
        }

        return result;
    }
876327
private Object receiveAndRetry(AmqpTemplate template) throws InterruptedException
    {
        Object message = null;
        try
        {
            message = template.receiveAndConvert();
        }
        catch (AmqpException ex)
        {
            // 接続断のタイミングにより、例外が出力される場合があるため、1秒後にリトライする。
            // 再度発生した例外は呼び出し元に返す。
            Thread.sleep(getRetryInterval());
            message = template.receiveAndConvert();
        }
        return message;
    }
543011
private synchronized void queueValid() throws InterruptedException {
        if(this.queue.isEmpty() && !this.faildQueue.isEmpty()){
            for (Task task : this.faildQueue) {
                this.push(task, false);
            }
            this.faildQueue.clear();
        }
        if (this.queue.isEmpty()) {
            logger.info(Thread.currentThread().getName() + " queue is empty");
        }
    }
1281538
public String read(String path, Watcher watcher, Stat stat) throws InterruptedException, KeeperException {

        return store.read(path, watcher, stat);
    }
855883
void doAwaitUntil() throws InterruptedException {
        lock.lock();
        try {
            condition.await(DEFAULT_TIMEOUT_MILLISECONDS, TimeUnit.MILLISECONDS);
        } finally {
            lock.unlock();
        }
    }
199721
public boolean enter() throws KeeperException, InterruptedException{
		zooKeeper.create(rootPath + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
		while (true) {
			synchronized (mutex) {
				List<String> list = zooKeeper.getChildren(rootPath, true);

				if (list.size() < size) {
					mutex.wait();
				} else {
					return true;
				}
			}
		}
	}
21920
@Override
    public void activate(final Object maintainer) throws InterruptedException, CouldNotPerformException {
        if (!isLocked() || this.maintainer.equals(maintainer)) {
            synchronized (maintainerLock) {
                unlock(maintainer);
                activate();
                lock(maintainer);
            }
        } else {
            throw new VerificationFailedException("[" + maintainer + "] is not the current maintainer of this remote");
        }
    }
1176493
@Override
    public void put(String key, JSONObject value) throws KeeperException, InterruptedException {
        try {
            m_zk.create(ZKUtil.joinZKPath(m_rootNode, key), value.toString().getBytes(Constants.UTF8ENCODING),
                    Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        } catch (KeeperException.NodeExistsException e) {
            m_zk.setData(ZKUtil.joinZKPath(m_rootNode, key), value.toString().getBytes(Constants.UTF8ENCODING), -1);
        }
    }
749294
@Override
	public final void cleanup(Context context) throws IOException, InterruptedException {
		cleanup(this.context, collector);
		collector.close();
		super.cleanup(context);
	}
789361
public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        // Note: convention in all put/take/etc is to preset local var
        // holding count negative to indicate failure unless set.
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            /*
             * Note that count is used in wait guard even though it is
             * not protected by lock. This works because count can
             * only decrease at this point (all other puts are shut
             * out by lock), and we (or some other waiting put) are
             * signalled if it ever changes from capacity. Similarly
             * for all other uses of count in other wait guards.
             */
            while (count.get() == capacity) {
                notFull.await();
            }
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }
1062451
private void startup() throws InterruptedException {
        try {
            arbitrateEventService.mainStemEvent().await(pipelineId);
        } catch (Throwable e) {
            if (isInterrupt(e)) {
                logger.info("INFO ## this node is interrupt", e);
            } else {
                logger.warn("WARN ## this node is crashed.", e);
            }
            arbitrateEventService.mainStemEvent().release(pipelineId);
            return;
        }

        executor = Executors.newFixedThreadPool(2); // 启动两个线程
        // 启动selector
        otterSelector = otterSelectorFactory.getSelector(pipelineId); // 获取对应的selector
        otterSelector.start();

        canStartSelector.set(false);// 初始化为false
        startProcessTermin();
        startProcessSelect();

        isStart = true;
    }
1254837
@Override
    public T take() throws InterruptedException {
        T old = poll();

        while (old == null) {
            waitGet_(-1);
            old = poll();
        }

        return old;
    }
910089
private boolean checkFailover() throws InterruptedException {
    boolean isPrimary = isPrimaryServiceCached();
    if (!isPrimary && isPrimaryService()) {
      this.servicePair.setPrimaryOfferService(this);
      return true;
    }
    return false;
  }
447563
public void invoke() throws ActionException {
		Integer position = interceptorPosition.get();
		if (position == null) {
			position = 0;
		}

		if (interceptors != null && position < interceptors.length) {
			//执行过滤器，递归调用本方法
			interceptorPosition.set(position + 1);
			interceptors[position].invoke(this);
		}else {
			//过滤器执行完毕，执行本体方法
			invokeActionMethod();
			// 执行了Action本体方法，说明过滤器使用完毕，清理游标防止重复执行
			resetInterceptorPosition();
		}
	}
1356628
private void manage() throws InterruptedException
  {
    log.info("Beginning management in %s.", config.getStartDelay());
    Thread.sleep(config.getStartDelay().getMillis());

    // Ignore return value- we'll get the IDs and futures from getKnownTasks later.
    taskRunner.restore();

    while (active) {
      giant.lock();

      try {
        // Task futures available from the taskRunner
        final Map<String, ListenableFuture<TaskStatus>> runnerTaskFutures = new HashMap<>();
        for (final TaskRunnerWorkItem workItem : taskRunner.getKnownTasks()) {
          runnerTaskFutures.put(workItem.getTaskId(), workItem.getResult());
        }
        // Attain futures for all active tasks (assuming they are ready to run).
        // Copy tasks list, as notifyStatus may modify it.
        for (final Task task : ImmutableList.copyOf(tasks)) {
          if (!taskFutures.containsKey(task.getId())) {
            final ListenableFuture<TaskStatus> runnerTaskFuture;
            if (runnerTaskFutures.containsKey(task.getId())) {
              runnerTaskFuture = runnerTaskFutures.get(task.getId());
            } else {
              // Task should be running, so run it.
              final boolean taskIsReady;
              try {
                taskIsReady = task.isReady(taskActionClientFactory.create(task));
              }
              catch (Exception e) {
                log.warn(e, "Exception thrown during isReady for task: %s", task.getId());
                notifyStatus(task, TaskStatus.failure(task.getId()), "failed because of exception[%s]", e.getClass());
                continue;
              }
              if (taskIsReady) {
                log.info("Asking taskRunner to run: %s", task.getId());
                runnerTaskFuture = taskRunner.run(task);
              } else {
                continue;
              }
            }
            taskFutures.put(task.getId(), attachCallbacks(task, runnerTaskFuture));
          } else if (isTaskPending(task)) {
            // if the taskFutures contain this task and this task is pending, also let the taskRunner
            // to run it to guarantee it will be assigned to run
            // see https://github.com/apache/incubator-druid/pull/6991
            taskRunner.run(task);
          }
        }
        // Kill tasks that shouldn't be running
        final Set<String> tasksToKill = Sets.difference(
            runnerTaskFutures.keySet(),
            ImmutableSet.copyOf(
                Lists.transform(
                    tasks,
                    new Function<Task, Object>()
                    {
                      @Override
                      public String apply(Task task)
                      {
                        return task.getId();
                      }
                    }
                )
            )
        );
        if (!tasksToKill.isEmpty()) {
          log.info("Asking taskRunner to clean up %,d tasks.", tasksToKill.size());
          for (final String taskId : tasksToKill) {
            try {
              taskRunner.shutdown(
                  taskId,
                  "task is not in runnerTaskFutures[%s]",
                  runnerTaskFutures.keySet()
              );
            }
            catch (Exception e) {
              log.warn(e, "TaskRunner failed to clean up task: %s", taskId);
            }
          }
        }
        // awaitNanos because management may become necessary without this condition signalling,
        // due to e.g. tasks becoming ready when other folks mess with the TaskLockbox.
        managementMayBeNecessary.awaitNanos(MANAGEMENT_WAIT_TIMEOUT_NANOS);
      }
      finally {
        giant.unlock();
      }
    }
  }
181067
@Override
    public void waitForData() throws CouldNotPerformException, InterruptedException {
        try {
            if (isDataAvailable()) {
                return;
            }
            logger.debug("Wait for " + this.toString() + " data...");
            getDataFuture().get();
            dataObservable.waitForValue();
        } catch (ExecutionException | CancellationException ex) {
            if (shutdownInitiated) {
                throw new InterruptedException("Interrupt request because system shutdown was initiated!");
            }
            throw new CouldNotPerformException("Could not wait for data!", ex);
        }
    }