com.google.common.util.concurrent.Uninterruptibles#getUninterruptibly ( )源码实例Demo

public static <V> V getDone(Future<V> future) throws ExecutionException {
  /*
   * We throw IllegalStateException, since the call could succeed later.
   * Perhaps we "should" throw IllegalArgumentException, since the call
   * could succeed with a different argument. Those exceptions' docs
   * suggest that either is acceptable. Google's Java Practices page
   * recommends IllegalArgumentException here, in part to keep its
   * recommendation simple: Static methods should throw
   * IllegalStateException only when they use static state.
   *
   *
   * Why do we deviate here? The answer: We want for fluentFuture.getDone()
    * to throw the same exception as Futures.getDone(fluentFuture).
   */
  Preconditions.checkState(future.isDone(), "Future was expected to be " +
      "done:" +
      " %s", future);
  return Uninterruptibles.getUninterruptibly(future);
}
 

/** Runs the entire J2CL pipeline. */
static Problems transpile(J2clTranspilerOptions options) {
  // Compiler has no static state, but rather uses thread local variables.
  // Because of this, we invoke the compiler on a different thread each time.
  ExecutorService executorService = Executors.newSingleThreadExecutor();
  Future<Problems> result =
      executorService.submit(() -> new J2clTranspiler(options).transpileImpl());
  // Shutdown the executor service since it will only run a single transpilation. If not shutdown
  // it prevents the JVM from ending the process (see Executors.newFixedThreadPool()). This is not
  // normally observed since the transpiler in normal circumstances ends with System.exit() which
  // ends all threads. But when the transpilation throws an exception, the exception propagates
  // out of main() and the process lingers due the live threads from these executors.
  executorService.shutdown();

  try {
    return Uninterruptibles.getUninterruptibly(result);
  } catch (ExecutionException e) {
    // Try unwrapping the cause...
    Throwables.throwIfUnchecked(e.getCause());
    throw new AssertionError(e.getCause());
  }
}
 

@Override
public void unlock() {
  if (!lock.isHeldByCurrentThread()) {
    throw new IllegalStateException("Cannot unlock without holding a lock by thread " + Thread.currentThread());
  }

  try {
    if (lock.getHoldCount() == 1) {
      // If it is the last lock entry for this thread, remove the zk node as well.
      try {
        Uninterruptibles.getUninterruptibly(zkClient.delete(localLockNode.get()));
      } catch (ExecutionException e) {
        throw Throwables.propagate(e.getCause());
      } finally {
        localLockNode.remove();
      }
    }
  } finally {
    lock.unlock();
  }
}
 

/**
 * Refreshes the value associated with {@code key}, unless another thread is already doing so.
 * Returns the newly refreshed value associated with {@code key} if it was refreshed inline, or
 * {@code null} if another thread is performing the refresh or if an error occurs during
 * refresh.
 */

@Nullable
V refresh(K key, int hash, CacheLoader<? super K, V> loader, boolean checkTime) {
  final LoadingValueReference<K, V> loadingValueReference = insertLoadingValueReference(key, hash, checkTime);
  if (loadingValueReference == null) {
    return null;
  }
  ListenableFuture<V> result = loadAsync(key, hash, loadingValueReference, loader);
  if (result.isDone()) {
    try {
      return Uninterruptibles.getUninterruptibly(result);
    } catch (Throwable t) {
      // don't let refresh exceptions propagate; error was already logged
    }
  }
  return null;
}
 

/**
 * Refreshes the value associated with {@code key}, unless another thread is already doing so.
 * Returns the newly refreshed value associated with {@code key} if it was refreshed inline, or
 * {@code null} if another thread is performing the refresh or if an error occurs during
 * refresh.
 */

@Nullable
V refresh(K key, int hash, CacheLoader<? super K, V> loader, boolean checkTime) {
  final LoadingValueReference<K, V> loadingValueReference = insertLoadingValueReference(key, hash, checkTime);
  if (loadingValueReference == null) {
    return null;
  }
  ListenableFuture<V> result = loadAsync(key, hash, loadingValueReference, loader);
  if (result.isDone()) {
    try {
      return Uninterruptibles.getUninterruptibly(result);
    } catch (Throwable t) {
      // don't let refresh exceptions propagate; error was already logged
    }
  }
  return null;
}
 

/**
 * Refreshes the value associated with {@code key}, unless another thread is already doing so.
 * Returns the newly refreshed value associated with {@code key} if it was refreshed inline, or
 * {@code null} if another thread is performing the refresh or if an error occurs during
 * refresh.
 */

@Nullable
V refresh(K key, int hash, CacheLoader<? super K, V> loader, boolean checkTime) {
  final LoadingValueReference<K, V> loadingValueReference = insertLoadingValueReference(key, hash, checkTime);
  if (loadingValueReference == null) {
    return null;
  }
  ListenableFuture<V> result = loadAsync(key, hash, loadingValueReference, loader);
  if (result.isDone()) {
    try {
      return Uninterruptibles.getUninterruptibly(result);
    } catch (Throwable t) {
      // don't let refresh exceptions propagate; error was already logged
    }
  }
  return null;
}
 

@Override
public void blazeShutdown() {
  if (closeFuturesWithTimeoutsMap.isEmpty()) {
    return;
  }

  try {
    Uninterruptibles.getUninterruptibly(
        Futures.allAsList(closeFuturesWithTimeoutsMap.values()),
        getMaxWaitForPreviousInvocation().getSeconds(),
        TimeUnit.SECONDS);
  } catch (TimeoutException | ExecutionException exception) {
    googleLogger.atWarning().withCause(exception).log(
        "Encountered Exception when closing BEP transports in Blaze's shutting down sequence");
  } finally {
    cancelAndResetPendingUploads();
  }
}
 

List<Path> globUninterruptible(
    Path base,
    Collection<String> patterns,
    boolean excludeDirectories,
    Predicate<Path> dirPred,
    FilesystemCalls syscalls)
    throws IOException, BadPattern {
  try {
    return Uninterruptibles.getUninterruptibly(
        globAsync(base, patterns, excludeDirectories, dirPred, syscalls));
  } catch (ExecutionException e) {
    Throwable cause = e.getCause();
    Throwables.propagateIfPossible(cause, IOException.class);
    Throwables.propagateIfPossible(cause, BadPattern.class);
    throw new RuntimeException(e);
  }
}
 

/**
 * Waits as long as necessary for all pending tasks to complete, performing shutdown operations if
 * necessary. When this method returns successfully, {@link #size()} {@code == 0}.
 *
 * @param terminate If the backing thread pool should be terminated once all tasks finish.
 * @throws ExecutionException If a pending task throws an exception.
 */
public void await(boolean terminate) throws ExecutionException {
    checkState(isRunning(), "Backing thread pool has already been terminated.");

    for (;;) {
        Future<?> pending = pendingTasks.poll();

        if (pending == null) {
            break;
        }

        Uninterruptibles.getUninterruptibly(pending);
    }

    if (terminate) {
        threadPool.shutdown();
        ThreadUtils.awaitTerminationUninterruptibly(threadPool);
    }
}
 

@Test
public void testRequestExpectsResponse() throws ExecutionException {
  HttpRequest httpRequest = Http.post("hostname.com", "/path", "this is the payload");
  Request request = client.writeExpectResponse(httpRequest);
  UUID id = Uninterruptibles.getUninterruptibly(request.getWriteFuture());

  ServerRequest serverRequest = Uninterruptibles.takeUninterruptibly(requests);
  HttpRequest serverHttpRequest = (HttpRequest) serverRequest.getPayload();
  assertEquals(id, serverRequest.getId());
  assertTrue(serverRequest.expectsResponse());
  assertEquals(httpRequest.method(), serverHttpRequest.method());
  assertEquals(httpRequest.uri(), serverHttpRequest.uri());
  assertEquals(httpRequest.protocolVersion(), serverHttpRequest.protocolVersion());
  assertEquals(httpRequest.headers().size(), serverHttpRequest.headers().size());
  for (Map.Entry<String, String> header : httpRequest.headers()) {
    assertTrue(serverHttpRequest.headers().contains(header.getKey(), header.getValue(), false));
  }

  Response response = Uninterruptibles.getUninterruptibly(request.getResponseFuture());
  assertEquals(request.getId(), response.getInResponseTo());
  HttpResponse responsePayload = (HttpResponse) response.getPayload();
}
 

private Object inputValueFromFuture(ListenableFuture<?> input) {
  try {
    return Uninterruptibles.getUninterruptibly(input);
  } catch (ExecutionException e) {
    Throwables.propagateIfInstanceOf(e.getCause(), GraphExecutionException.class);
    throw Throwables.propagate(e);
  }
}
 

private void forceShutdownBuildEventStreamer() {
  streamer.close(AbortReason.INTERNAL);
  closeFuturesWithTimeoutsMap =
      constructCloseFuturesMapWithTimeouts(streamer.getCloseFuturesMap());
  try {
    googleLogger.atInfo().log("Closing pending build event transports");
    Uninterruptibles.getUninterruptibly(Futures.allAsList(closeFuturesWithTimeoutsMap.values()));
  } catch (ExecutionException e) {
    googleLogger.atSevere().withCause(e).log("Failed to close a build event transport");
  } finally {
    cancelAndResetPendingUploads();
  }
}
 

protected void internalStop() {
  if (leaderElection != null) {
    // NOTE: if was a leader this will cause loosing of leadership which in callback above will
    //       de-register service in discovery service and stop the service if needed
    try {
      Uninterruptibles.getUninterruptibly(leaderElection.stop(), 5, TimeUnit.SECONDS);
    } catch (TimeoutException te) {
      LOG.warn("Timed out waiting for leader election cancellation to complete");
    } catch (ExecutionException e) {
      LOG.error("Exception when cancelling leader election.", e);
    }
  }
}
 

/**
 * Returns the current data related to the instance from {@link #performRefresh()}. May block if
 * no valid data is currently available.
 */
private InstanceData getInstanceData() {
  ListenableFuture<InstanceData> instanceData;
  synchronized (instanceDataGuard) {
    instanceData = currentInstanceData;
  }
  try {
    // TODO(kvg): Let exceptions up to here before adding context
    return Uninterruptibles.getUninterruptibly(instanceData);
  } catch (ExecutionException ex) {
    Throwable cause = ex.getCause();
    Throwables.throwIfUnchecked(cause);
    throw new RuntimeException(cause);
  }
}
 

private void close() throws IOException {
  LOG.info("SSTables built. Now starting streaming");
  heartbeat.startHeartbeat();
  try {
    if (writer != null) {
      writer.close();
      Future<StreamState> future =
          loader.stream(Collections.<InetAddress>emptySet(), new ProgressIndicator());
      try {
        StreamState streamState = Uninterruptibles.getUninterruptibly(future);
        if (streamState.hasFailedSession()) {
          LOG.warn("Some streaming sessions failed");
        } else {
          LOG.info("Streaming finished successfully");
        }
      } catch (ExecutionException e) {
        throw new RuntimeException("Streaming to the following hosts failed: " +
            loader.getFailedHosts(), e);
      }
    } else {
      LOG.info("SSTableWriter wasn't instantiated, no streaming happened.");
    }
  } finally {
    heartbeat.stopHeartbeat();
  }
  LOG.info("Successfully closed bulk record writer");
}
 

@Override
public void waitForCompletion() throws IOException {
  try {
    Uninterruptibles.getUninterruptibly(future);
  } catch (ExecutionException ee) {
    // Runnable threw a RuntimeException
    Throwable nested = ee.getCause();
    if (nested instanceof RuntimeException) {
      final RuntimeException re = (RuntimeException) nested;
      // The stream sink classes, unfortunately, tunnel IOExceptions
      // out of run() in a RuntimeException. If that's the case,
      // unpack and re-throw the IOException. Otherwise, re-throw
      // this unexpected RuntimeException
      final Throwable cause = re.getCause();
      if (cause instanceof IOException) {
        throw (IOException) cause;
      } else {
        throw re;
      }
    } else if (nested instanceof OutOfMemoryError) {
      // OutOfMemoryError does not support exception chaining.
      throw (OutOfMemoryError) nested;
    } else if (nested instanceof Error) {
      throw new Error("unhandled Error in worker thread", ee);
    } else {
      throw new RuntimeException("unknown execution problem", ee);
    }
  }
}
 

@Override
public void service(final HttpServletRequest req, final HttpServletResponse rsp)
    throws ServletException, IOException {
  FutureTask<Void> task = new FutureTask<>(new Callable<Void>() {
    @Nullable
    @Override
    public Void call() throws ServletException, IOException {
      // Simulate the full filter chain with the servlet at the end.
      final Iterator<Class<? extends Filter>> filtersIter = filterClasses.iterator();
      FilterChain filterChain =
          new FilterChain() {
            @Override
            public void doFilter(ServletRequest request, ServletResponse response)
                throws IOException, ServletException {
              if (filtersIter.hasNext()) {
                instantiate(filtersIter.next()).doFilter(request, response, this);
              } else {
                instantiate(servletClass).service(request, response);
              }
            }};
      filterChain.doFilter(req, rsp);
      return null;
    }});
  requestQueue.add(task);
  try {
    Uninterruptibles.getUninterruptibly(task);
  } catch (ExecutionException e) {
    throwIfInstanceOf(e.getCause(), ServletException.class);
    throwIfInstanceOf(e.getCause(), IOException.class);
    throw new RuntimeException(e.getCause());
  }
}
 

@Override
public void awaitTerminated() throws ExecutionException {
  Uninterruptibles.getUninterruptibly(terminationFuture);
}
 

@Override
public void awaitTerminated(long timeout, TimeUnit timeoutUnit) throws TimeoutException, ExecutionException {
  Uninterruptibles.getUninterruptibly(terminationFuture, timeout, timeoutUnit);
}
 

private void waitForBuildEventTransportsToClose(
    Map<BuildEventTransport, ListenableFuture<Void>> transportFutures,
    boolean besUploadModeIsSynchronous)
    throws AbruptExitException {
  final ScheduledExecutorService executor =
      Executors.newSingleThreadScheduledExecutor(
          new ThreadFactoryBuilder().setNameFormat("bes-notify-ui-%d").build());
  ScheduledFuture<?> waitMessageFuture = null;

  try {
    // Notify the UI handler when a transport finished closing.
    transportFutures.forEach(
        (bepTransport, closeFuture) ->
            closeFuture.addListener(
                () -> {
                  reporter.post(new BuildEventTransportClosedEvent(bepTransport));
                },
                executor));

    try (AutoProfiler p = GoogleAutoProfilerUtils.logged("waiting for BES close")) {
      Uninterruptibles.getUninterruptibly(Futures.allAsList(transportFutures.values()));
    }
  } catch (ExecutionException e) {
    // Futures.withTimeout wraps the TimeoutException in an ExecutionException when the future
    // times out.
    if (isTimeoutException(e)) {
      throw createAbruptExitException(
          e,
          "The Build Event Protocol upload timed out.",
          ExitCode.TRANSIENT_BUILD_EVENT_SERVICE_UPLOAD_ERROR,
          BuildProgress.Code.BES_UPLOAD_TIMEOUT_ERROR);
    }

    Throwables.throwIfInstanceOf(e.getCause(), AbruptExitException.class);
    throw new RuntimeException(
        String.format(
            "Unexpected Exception '%s' when closing BEP transports, this is a bug.",
            e.getCause().getMessage()));
  } finally {
    if (besUploadModeIsSynchronous) {
      cancelAndResetPendingUploads();
    }
    if (waitMessageFuture != null) {
      waitMessageFuture.cancel(/* mayInterruptIfRunning= */ true);
    }
    executor.shutdown();
  }
}