java.util.concurrent.ConcurrentLinkedQueue#peek ( )源码实例Demo

/**
 * Cleans up the {@link SnackBarItem} and the {@link Activity} it is tied to
 *
 * @param activity     The {@link Activity} tied to the {@link SnackBarItem}
 * @param snackBarItem The {@link SnackBarItem} to clean up
 */
public void disposeSnackBar(Activity activity, SnackBarItem snackBarItem) {
    ConcurrentLinkedQueue<SnackBarItem> list = mQueue.get(activity);

    if (list != null) {
        list.remove(snackBarItem);

        if (list.peek() == null) {
            mQueue.remove(activity);
            mIsShowingSnackBar = false;
        } else if (!mIsCanceling) {
            mIsShowingSnackBar = true;
            list.peek().show();
        }
    }
}
 

@Test
public void givenProducerOffersElementInQueue_WhenConsumerPollsQueue_ThenItRetrievesElement() throws Exception {
    int element = 1;

    ExecutorService executorService = Executors.newFixedThreadPool(2);
    ConcurrentLinkedQueue<Integer> concurrentLinkedQueue = new ConcurrentLinkedQueue<>();
    Runnable offerTask = () -> concurrentLinkedQueue.offer(element);

    Callable<Integer> pollTask = () -> {
        while (concurrentLinkedQueue.peek() != null) {
            return concurrentLinkedQueue.poll()
                .intValue();
        }
        return null;
    };

    executorService.submit(offerTask);
    TimeUnit.SECONDS.sleep(1);

    Future<Integer> returnedElement = executorService.submit(pollTask);
    assertThat(returnedElement.get()
        .intValue(), is(equalTo(element)));
    executorService.awaitTermination(1, TimeUnit.SECONDS);
    executorService.shutdown();
}
 

public boolean awaitCompletion(long timeout, TimeUnit unit) throws InterruptedException {
  if ( unit == null ) throw new IllegalArgumentException("unit cannot be null");
  // get a snapshot so we only look at tasks already queued, not any that
  // get asynchronously queued after this point
  ConcurrentLinkedQueue<Runnable> snapshotQueuedAndExecutingTasks = snapshotQueuedAndExecutingTasks();
  try {
    long duration = unit.convert(timeout, TimeUnit.MILLISECONDS);
    // we can iterate even when the underlying set is being modified
    // since we're using ConcurrentHashMap
    Runnable task = null;
    while((task = snapshotQueuedAndExecutingTasks.peek()) != null) {
      // Lock task before we re-check whether it is queued or executing in
      // the main set and in the snapshot.  Thus there's no way for the
      // notifyAll to sneak in right after our check and leave us waiting
      // forever.  Also we already have the lock required to call
      // task.wait().  Normally we religiously avoid any synchronized
      // blocks, but we couldn't find a way to avoid this one.
      synchronized(task) {
        while ( snapshotQueuedAndExecutingTasks.contains(task) &&
          queuedAndExecutingTasks.contains(task) )
        {
          long startTime = System.currentTimeMillis();
          // block until task is complete or timeout expires
          task.wait(duration);
          duration -= System.currentTimeMillis() - startTime;
          if ( duration <= 0 ) {
            // times up!  We didn't finish before timeout...
            logger.debug("[awaitCompletion] timeout");
            return false;
          }
        }
      }
    }
  } finally {
    removeSnapshot();
  }
  return true;
}