本文整理了Java中java.util.concurrent.BlockingQueue类的一些代码示例，展示了BlockingQueue类的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。BlockingQueue类的具体详情如下：
包路径：java.util.concurrent.BlockingQueue
类名称：BlockingQueue

BlockingQueue介绍

[英]A java.util.Queue that additionally supports operations that wait for the queue to become non-empty when retrieving an element, and wait for space to become available in the queue when storing an element.

BlockingQueue methods come in four forms, with different ways of handling operations that cannot be satisfied immediately, but may be satisfied at some point in the future: one throws an exception, the second returns a special value (either null or false, depending on the operation), the third blocks the current thread indefinitely until the operation can succeed, and the fourth blocks for only a given maximum time limit before giving up. These methods are summarized in the following table:

Throws exceptionSpecial valueBlocks**Times outInsert#add#offer#put#offer(Object,long,TimeUnit)Remove#remove#poll#take#poll(long,TimeUnit)Examine#element#peeknot applicable**not applicable

A BlockingQueue does not accept null elements. Implementations throw NullPointerException on attempts to add, put or offer a null. A null is used as a sentinel value to indicate failure of poll operations.

A BlockingQueue may be capacity bounded. At any given time it may have a remainingCapacity beyond which no additional elements can be put without blocking. A BlockingQueue without any intrinsic capacity constraints always reports a remaining capacity of Integer.MAX_VALUE.

BlockingQueue implementations are designed to be used primarily for producer-consumer queues, but additionally support the java.util.Collection interface. So, for example, it is possible to remove an arbitrary element from a queue using remove(x). However, such operations are in general not performed very efficiently, and are intended for only occasional use, such as when a queued message is cancelled.

BlockingQueue implementations are thread-safe. All queuing methods achieve their effects atomically using internal locks or other forms of concurrency control. However, the bulk Collection operations addAll, containsAll, retainAll and removeAll are not necessarily performed atomically unless specified otherwise in an implementation. So it is possible, for example, for addAll(c) to fail (throwing an exception) after adding only some of the elements in c.

A BlockingQueue does not intrinsically support any kind of "close" or "shutdown" operation to indicate that no more items will be added. The needs and usage of such features tend to be implementation-dependent. For example, a common tactic is for producers to insert special end-of-stream or poison objects, that are interpreted accordingly when taken by consumers.

Usage example, based on a typical producer-consumer scenario. Note that a BlockingQueue can safely be used with multiple producers and multiple consumers.

class Producer implements Runnable { 
private final BlockingQueue queue; 
Producer(BlockingQueue q) { queue = q; } 
public void run() { 
try { 
while (true) { queue.put(produce()); } 
} catch (InterruptedException ex) { ... handle ...} 
} 
Object produce() { ... } 
} 
class Consumer implements Runnable { 
private final BlockingQueue queue; 
Consumer(BlockingQueue q) { queue = q; } 
public void run() { 
try { 
while (true) { consume(queue.take()); } 
} catch (InterruptedException ex) { ... handle ...} 
} 
void consume(Object x) { ... } 
} 
class Setup { 
void main() { 
BlockingQueue q = new SomeQueueImplementation(); 
Producer p = new Producer(q); 
Consumer c1 = new Consumer(q); 
Consumer c2 = new Consumer(q); 
new Thread(p).start(); 
new Thread(c1).start(); 
new Thread(c2).start(); 
} 
}

Memory consistency effects: As with other concurrent collections, actions in a thread prior to placing an object into a BlockingQueuehappen-before actions subsequent to the access or removal of that element from the BlockingQueue in another thread.

This interface is a member of the Java Collections Framework.
[中]爪哇语。util。该队列还支持以下操作：检索元素时等待队列变为非空，存储元素时等待队列中的空间变为可用。
BlockingQueue方法有四种形式，有不同的处理操作的方法，这些操作不能立即满足，但在将来的某个时候可能会满足：一种抛出异常，另一种返回特殊值（null或false，取决于操作），第三个线程无限期地阻塞当前线程，直到操作成功，第四个线程在放弃之前只阻塞给定的最大时间限制。下表总结了这些方法：
抛出异常特殊值块**超时Insert#add#offer#put#offer（Object，long，TimeUnit）Remove#Remove#poll#take#poll（long，TimeUnit）Inspect#元素#peek不适用**不适用
BlockingQueue不接受空元素。实现在尝试添加、放置或提供null时抛出NullPointerException。null用作sentinel值，以指示轮询操作失败。
阻塞队列可能有容量限制。在任何给定的时间，它都可能有一个剩余容量，超过这个容量，就不能不阻塞地放置额外的元件。没有任何内在容量约束的BlockingQueue始终报告整数的剩余容量。最大值。
BlockingQueue实现主要用于生产者-消费者队列，但还支持java。util。收集接口。因此，例如，可以使用remove（x）从队列中删除任意元素。然而，此类操作通常不是非常有效地执行，并且仅用于偶尔使用，例如当排队消息被取消时。
BlockingQueue实现是线程安全的。所有排队方法都使用内部锁或其他形式的并发控制以原子方式实现其效果。但是，除非在实现中另有规定，否则批量收集操作addAll、containsAll、retainal和removeAll不必以原子方式执行。因此，例如，addAll（c）在只添加c中的一些元素之后失败（引发异常）是可能的。
BlockingQueue不本质上支持任何类型的“关闭”或“关闭”操作，以指示不再添加任何项目。这些特性的需求和使用往往取决于实现。例如，一种常见的策略是生产者插入特殊的流结束或毒药对象，消费者使用这些对象时会相应地进行解释。
使用示例，基于典型的生产者-消费者场景。请注意，BlockingQueue可以安全地用于多个生产者和多个消费者。

class Producer implements Runnable { 
private final BlockingQueue queue; 
Producer(BlockingQueue q) { queue = q; } 
public void run() { 
try { 
while (true) { queue.put(produce()); } 
} catch (InterruptedException ex) { ... handle ...} 
} 
Object produce() { ... } 
} 
class Consumer implements Runnable { 
private final BlockingQueue queue; 
Consumer(BlockingQueue q) { queue = q; } 
public void run() { 
try { 
while (true) { consume(queue.take()); } 
} catch (InterruptedException ex) { ... handle ...} 
} 
void consume(Object x) { ... } 
} 
class Setup { 
void main() { 
BlockingQueue q = new SomeQueueImplementation(); 
Producer p = new Producer(q); 
Consumer c1 = new Consumer(q); 
Consumer c2 = new Consumer(q); 
new Thread(p).start(); 
new Thread(c1).start(); 
new Thread(c2).start(); 
} 
}

内存一致性影响：与其他并发集合一样，在将对象放入BlockingQueuehappen-before之前线程中的操作在另一个线程中从BlockingQueue访问或移除该元素之后的操作。
此接口是Java Collections Framework的成员。

代码示例

代码示例来源：origin: square/okhttp

@Override public MockResponse dispatch(RecordedRequest request) throws InterruptedException {
 // To permit interactive/browser testing, ignore requests for favicons.
 final String requestLine = request.getRequestLine();
 if (requestLine != null && requestLine.equals("GET /favicon.ico HTTP/1.1")) {
  logger.info("served " + requestLine);
  return new MockResponse().setResponseCode(HttpURLConnection.HTTP_NOT_FOUND);
 }
 if (failFastResponse != null && responseQueue.peek() == null) {
  // Fail fast if there's no response queued up.
  return failFastResponse;
 }
 MockResponse result = responseQueue.take();
 // If take() returned because we're shutting down, then enqueue another dead letter so that any
 // other threads waiting on take() will also return.
 if (result == DEAD_LETTER) responseQueue.add(DEAD_LETTER);
 return result;
}

代码示例来源：origin: code4craft/webmagic

@Override
public synchronized Request poll(Task task) {
  if (!inited.get()) {
    init(task);
  }
  fileCursorWriter.println(cursor.incrementAndGet());
  return queue.poll();
}

代码示例来源：origin: apache/incubator-druid

@Override
 public void rejectedExecution(Runnable r, ThreadPoolExecutor executor)
 {
  try {
   executor.getQueue().put(r);
  }
  catch (InterruptedException e) {
   throw new RejectedExecutionException("Got Interrupted while adding to the Queue", e);
  }
 }
}

代码示例来源：origin: ReactiveX/RxJava

@Override
public void onNext(Notification<T> args) {
  if (waiting.getAndSet(0) == 1 || !args.isOnNext()) {
    Notification<T> toOffer = args;
    while (!buf.offer(toOffer)) {
      Notification<T> concurrentItem = buf.poll();
      // in case if we won race condition with onComplete/onError method
      if (concurrentItem != null && !concurrentItem.isOnNext()) {
        toOffer = concurrentItem;
      }
    }
  }
}

代码示例来源：origin: apache/nifi

if (this.closed.get()) {
  logger.info("Provenance Repository has been closed; will not merge journal files to {}", suggestedMergeFile);
  return null;
    final AtomicBoolean finishedAdding = new AtomicBoolean(false);
    final List<Future<?>> futures = new ArrayList<>();
      final AtomicInteger indexingFailureCount = new AtomicInteger(0);
      try {
        for (int i = 0; i < configuration.getIndexThreadPoolSize(); i++) {
          final Future<?> future = exec.submit(callable);
          futures.add(future);
          while (!accepted && indexEvents) {
            try {
              accepted = eventQueue.offer(new Tuple<>(record, blockIndex), 10, TimeUnit.MILLISECONDS);
            } catch (final InterruptedException ie) {
              Thread.currentThread().interrupt();
            if (!accepted && indexingFailureCount.get() >= MAX_INDEXING_FAILURE_COUNT) {
              indexEvents = false;  // don't add anything else to the queue.
              eventQueue.clear();
        finishedAdding.set(true);
        exec.shutdown();

代码示例来源：origin: apache/nifi

public void setup() {
  // Create a single script engine, the Processor object is reused by each task
  if(scriptEngine == null) {
    scriptingComponentHelper.setup(1, getLogger());
    scriptEngine = scriptingComponentHelper.engineQ.poll();
  }
  if (scriptEngine == null) {
    throw new ProcessException("No script engine available!");
  }
  if (scriptNeedsReload.get() || processor.get() == null) {
    if (ScriptingComponentHelper.isFile(scriptingComponentHelper.getScriptPath())) {
      reloadScriptFile(scriptingComponentHelper.getScriptPath());
    } else {
      reloadScriptBody(scriptingComponentHelper.getScriptBody());
    }
    scriptNeedsReload.set(false);
  }
}

代码示例来源：origin: Atmosphere/atmosphere

protected void dispatchMessages(Deliver e) {
  messages.offer(e);
  if (dispatchThread.get() == 0) {
    dispatchThread.incrementAndGet();
    getBroadcasterConfig().getExecutorService().submit(getBroadcastHandler());
  }
}

代码示例来源：origin: debezium/debezium

@Override
public List<SourceRecord> poll() throws InterruptedException {
  failureException = this.failure.get();
  if (failureException != null) {
  if (!running.get()) {
    cleanupResources();
    throw new InterruptedException( "Reader was stopped while polling" );
  List<SourceRecord> batch = new ArrayList<>(maxBatchSize);
  final Timer timeout = Threads.timer(Clock.SYSTEM, Temporals.max(pollInterval, ConfigurationDefaults.RETURN_CONTROL_INTERVAL));
  while (running.get() && (records.drainTo(batch, maxBatchSize) == 0) && !success.get()) {
    failureException = this.failure.get();
    if (failureException != null) throw failureException;
    if (timeout.expired()) {
  if (batch.isEmpty() && success.get() && records.isEmpty()) {
    this.running.set(false);

代码示例来源：origin: Atmosphere/atmosphere

Deliver msg = null;
try {
  msg = messages.poll(waitTime, TimeUnit.MILLISECONDS);
  if (msg == null) {
    dispatchThread.decrementAndGet();
    return;
  return;
} finally {
  if (outOfOrderBroadcastSupported.get()) {
    bc.getExecutorService().submit(this);
  push(msg);
} catch (Throwable ex) {
  if (!started.get() || destroyed.get()) {
    logger.trace("Failed to submit broadcast handler runnable on shutdown for Broadcaster {}", getID(), ex);
    return;

代码示例来源：origin: Netflix/eureka

getDeltaCount.getAndIncrement();
  if (sentDelta.compareAndSet(false, true)) {
    addDeltaApps(includeRemote, apps);
  } else {
  handled = true;
} else if (pathInfo.equals("apps/")) {
  getFullRegistryCount.getAndIncrement();
  if (sentDelta.get()) {
    addDeltaApps(includeRemote, apps);
  } else {
  sentRegistry.set(true);
  handled = true;
} else if (pathInfo.startsWith("vips/")) {
  getSingleVipCount.getAndIncrement();
    registrationStatusesQueue.add(statusStr);
    registrationStatuses.add(statusStr);

代码示例来源：origin: apache/incubator-gobblin

public void pushToStream(String message) {
 int streamNo = (int) this.nextStream.incrementAndGet() % this.queues.size();
 AtomicLong offset = this.offsets.get(streamNo);
 BlockingQueue<FetchedDataChunk> queue = this.queues.get(streamNo);
 AtomicLong thisOffset = new AtomicLong(offset.incrementAndGet());
 List<Message> seq = Lists.newArrayList();
 seq.add(new Message(message.getBytes(Charsets.UTF_8)));
 ByteBufferMessageSet messageSet = new ByteBufferMessageSet(NoCompressionCodec$.MODULE$, offset, JavaConversions.asScalaBuffer(seq));
 FetchedDataChunk chunk = new FetchedDataChunk(messageSet,
   new PartitionTopicInfo("topic", streamNo, queue, thisOffset, thisOffset, new AtomicInteger(1), "clientId"),
   thisOffset.get());
 queue.add(chunk);
}

代码示例来源：origin: apache/nifi

@Override
  public Object call() throws IOException {
    while (!eventQueue.isEmpty() || !finishedAdding.get()) {
      try {
        final Tuple<StandardProvenanceEventRecord, Integer> tuple;
        try {
          tuple = eventQueue.poll(10, TimeUnit.MILLISECONDS);
        } catch (final InterruptedException ie) {
          Thread.currentThread().interrupt();
          continue;
        }
        if (tuple == null) {
          continue;
        }
        indexingAction.index(tuple.getKey(), indexWriter.getIndexWriter(), tuple.getValue());
      } catch (final Throwable t) {
        logger.error("Failed to index Provenance Event for " + writerFile + " to " + indexingDirectory, t);
        if (indexingFailureCount.incrementAndGet() >= MAX_INDEXING_FAILURE_COUNT) {
          return null;
        }
      }
    }
    return null;
  }
};

代码示例来源：origin: robovm/robovm

int c = ctl.get();
if (isRunning(c) ||
  runStateAtLeast(c, TIDYING) ||
  (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
  return;
if (workerCountOf(c) != 0) { // Eligible to terminate
mainLock.lock();
try {
  if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
    try {
      terminated();
    } finally {
      ctl.set(ctlOf(TERMINATED, 0));
      termination.signalAll();
  mainLock.unlock();

代码示例来源：origin: apache/nifi

final SocketChannel socketChannel = channel.accept();
      if (currentConnections.incrementAndGet() > maxConnections){
        currentConnections.decrementAndGet();
        logger.warn("Rejecting connection from {} because max connections has been met",
            new Object[]{ socketChannel.getRemoteAddress().toString() });
      ByteBuffer buffer = bufferPool.poll();
      buffer.clear();
      buffer.mark();
      executor.execute(handler);
while((key = keyQueue.poll()) != null){
  key.interestOps(SelectionKey.OP_READ);

代码示例来源：origin: prestodb/presto

public BufferResult getPages(long sequenceId, DataSize maxSize)
  if (completed.get() && serializedPages.isEmpty()) {
    return BufferResult.emptyResults(TASK_INSTANCE_ID, token.get(), true);
  assertEquals(sequenceId, token.get(), "token");
    serializedPage = serializedPages.poll(10, TimeUnit.MILLISECONDS);
    return BufferResult.emptyResults(TASK_INSTANCE_ID, token.get(), false);
  long responseSize = serializedPage.getSizeInBytes();
  while (responseSize < maxSize.toBytes()) {
    serializedPage = serializedPages.poll();
    if (serializedPage == null) {
      break;

代码示例来源：origin: apache/incubator-druid

@Override
public ClientResponse<InputStream> done(ClientResponse<InputStream> clientResponse)
{
 synchronized (done) {
  try {
   // An empty byte array is put at the end to give the SequenceInputStream.close() as something to close out
   // after done is set to true, regardless of the rest of the stream's state.
   queue.put(ByteSource.empty().openStream());
   log.debug("Added terminal empty stream");
  }
  catch (InterruptedException e) {
   log.warn(e, "Thread interrupted while adding to queue");
   Thread.currentThread().interrupt();
   throw Throwables.propagate(e);
  }
  catch (IOException e) {
   // This should never happen
   log.wtf(e, "The empty stream threw an IOException");
   throw Throwables.propagate(e);
  }
  finally {
   log.debug("Done after adding %d bytes of streams", byteCount.get());
   done.set(true);
  }
 }
 return ClientResponse.finished(clientResponse.getObj());
}

代码示例来源：origin: apache/incubator-dubbo

final AtomicInteger count = new AtomicInteger();
final BlockingQueue<Object> ref = new LinkedBlockingQueue<>();
for (final Invoker<T> invoker : selected) {
  executor.execute(new Runnable() {
    @Override
    public void run() {
  Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
  if (ret instanceof Throwable) {
    Throwable e = (Throwable) ret;

代码示例来源：origin: actiontech/dble

public boolean offer(ServerConnection con, String nextStep, RouteResultset rrs) {
  PauseTask task = new PauseTask(rrs, nextStep, con);
  queueLock.lock();
  try {
    if (!teminateFlag) {
      if (queueNumber.incrementAndGet() <= queueLimit) {
        handlerQueue.offer(task);
        return true;
      } else {
        con.writeErrMessage(ER_YES, "The node is pausing, wait list is full");
        queueNumber.decrementAndGet();
      }
      return true;
    } else {
      return false;
    }
  } finally {
    queueLock.unlock();
  }
}

代码示例来源：origin: apache/incubator-druid

@Override
public void exceptionCaught(final ClientResponse<InputStream> clientResponse, final Throwable e)
{
 // Don't wait for lock in case the lock had something to do with the error
 synchronized (done) {
  done.set(true);
  // Make a best effort to put a zero length buffer into the queue in case something is waiting on the take()
  // If nothing is waiting on take(), this will be closed out anyways.
  final boolean accepted = queue.offer(
    new InputStream()
    {
     @Override
     public int read() throws IOException
     {
      throw new IOException(e);
     }
    }
  );
  if (!accepted) {
   log.warn("Unable to place final IOException offer in queue");
  } else {
   log.debug("Placed IOException in queue");
  }
  log.debug(e, "Exception with queue length of %d and %d bytes available", queue.size(), byteCount.get());
 }
}

代码示例来源：origin: apache/hive

Long lastKillTimeMs = null;
SanityChecker sc = null;
while (!isShutdown.get()) {
 RejectedExecutionException rejectedException = null;
 if (nextSanityCheck != null && ((nextSanityCheck - System.nanoTime()) <= 0)) {
  boolean shouldWait = numSlotsAvailable.get() <= 0 && lastKillTimeMs == null;
  boolean canKill = false;
  if (task.canFinishForPriority() || task.isGuaranteed()) {
      + "preemptionQueueSize={}, numSlotsAvailable={}, waitQueueSize={}",
      task.getRequestId(), task.getTaskRunnerCallable().canFinish(),
      preemptionQueue.size(), numSlotsAvailable.get(), waitQueue.size());
   canKill = enablePreemption && canPreempt(task, preemptionQueue.peek());
   shouldWait = shouldWait && !canKill;
if (isShutdown.get()) {
 LOG.info(WAIT_QUEUE_SCHEDULER_THREAD_NAME_FORMAT
   + " thread has been interrupted after shutdown.");