Thread-safe call to get a shared {@link BlockingQueue}.
772780
private BlockingQueue<V> retrieveSharedQueue(String key) {
		BlockingQueue<V> queue = mediations.get(key);
		if (queue == null) {
			queue = new ArrayBlockingQueue<V>(1);
			BlockingQueue<V> commonQueue = mediations.putIfAbsent(key, queue);
			return commonQueue != null ? commonQueue : queue;
		} else {
			return queue;
		}
	}
854316
private ClientRWLock getBlockLock(long blockId) {
    // Loop until we either find the block lock in the mLocks map, or successfully acquire a new
    // block lock from the lock pool.
    while (true) {
      ClientRWLock blockLock;
      // Check whether a lock has already been allocated for the block id.
      synchronized (mSharedMapsLock) {
        blockLock = mLocks.get(blockId);
        if (blockLock != null) {
          blockLock.addReference();
          return blockLock;
        }
      }
      // Since a block lock hasn't already been allocated, try to acquire a new one from the pool.
      // Acquire the lock outside the synchronized section because #acquire might need to block.
      // We shouldn't wait indefinitely in acquire because the another lock for this block could be
      // allocated to another thread, in which case we could just use that lock.
      blockLock = mLockPool.acquire(1, TimeUnit.SECONDS);
      if (blockLock != null) {
        synchronized (mSharedMapsLock) {
          // Check if someone else acquired a block lock for blockId while we were acquiring one.
          if (mLocks.containsKey(blockId)) {
            mLockPool.release(blockLock);
            blockLock = mLocks.get(blockId);
          } else {
            mLocks.put(blockId, blockLock);
          }
          blockLock.addReference();
          return blockLock;
        }
      }
    }
  }
395653
public int getRunningThreadCount() {
        int rc = 0;
        WorkQueue[] ws; WorkQueue w;
        if ((ws = workQueues) != null) {
            for (int i = 1; i < ws.length; i += 2) {
                if ((w = ws[i]) != null && w.isApparentlyUnblocked())
                    ++rc;
            }
        }
        return rc;
    }
128538
public static BlockingQueue<Runnable> buildQueue(int size, boolean isPriority) {
        BlockingQueue<Runnable> queue;
        if (size == 0) { // 默认无队列
            queue = new SynchronousQueue<Runnable>();
        } else { // 有限队列或无限队列
            if (isPriority) {
                queue = size < 0 ? new PriorityBlockingQueue<Runnable>()
                    : new PriorityBlockingQueue<Runnable>(size);
            } else {
                queue = size < 0 ? new LinkedBlockingQueue<Runnable>()
                    : new LinkedBlockingQueue<Runnable>(size);
            }
        }
        return queue;
    }
709112
private static <T> ListenableFuture<T> submitAndAddQueueListener(
      ListeningExecutorService executorService, Callable<T> task,
      final BlockingQueue<Future<T>> queue) {
    final ListenableFuture<T> future = executorService.submit(task);
    future.addListener(new Runnable() {
      @Override public void run() {
        queue.add(future);
      }
    }, directExecutor());
    return future;
  }
401656
public T    getQueue()
    {
        Preconditions.checkState(state.get() == State.STARTED, "Not started");

        List<String>    localPreferredQueues = Lists.newArrayList(preferredQueues);
        if ( localPreferredQueues.size() > 0 )
        {
            String      key = localPreferredQueues.get(random.nextInt(localPreferredQueues.size()));
            return queues.get(key);
        }

        List<String>    keys = Lists.newArrayList(queues.keySet());
        String          key = keys.get(random.nextInt(keys.size()));
        return queues.get(key);
    }
1131265
public boolean lock(boolean shared, boolean block)
  {
    if (shared && !_canRead) {
      // Invalid request for a shared "read" lock on a write only stream.

      return false;
    }

    if (!shared && !_canWrite) {
      // Invalid request for an exclusive "write" lock on a read only stream.

      return false;
    }

    return true;
  }
768608
@Override
    public java.util.concurrent.Future<GetQueueUrlResult> getQueueUrlAsync(String queueName) {

        return getQueueUrlAsync(new GetQueueUrlRequest().withQueueName(queueName));
    }
665800
public void dispatchAsynchException(ProxyQueue proxyQueue, Exception exception)
   {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(tc, "dispatchAsynchException",
                                           new Object[] { proxyQueue, exception });

      // Create a runnable with the data
      AsynchExceptionThread thread = new AsynchExceptionThread(proxyQueue, exception);
      dispatchThread(thread);

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "dispatchAsynchException");
   }
412300
@Override
    public V poll(long timeout, TimeUnit unit) throws InterruptedException {
        return blockingQueue.poll(timeout, unit);
    }
20916
public static <V> StrategicBlockingQueue<V> newStrategicLinkedBlockingQueue(QueueingStrategy<V> queueingStrategy) {
        return new StrategicBlockingQueue<V>(new LinkedBlockingQueue<V>(), queueingStrategy);
    }
346763
private SortedQueue getQueueReadOnly(String queueName, Duration waitDuration) {
        DedupQueue service = getQueueReadWrite(queueName, waitDuration);
        if (service != null) {
            try {
                return service.getQueue();
            } catch (ReadOnlyQueueException e) {
                // Fall through
            }
        }
        return _sortedQueueFactory.create(queueName, true, _queueDAO);
    }
32097
protected void enqueue (final AbstractInvocation invocation) {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "enqueue", invocation);

    barrier.pass(); // Block until allowed to pass
  
    // We need to ensure that the thread processing this queue does not change the "running" state before we decide whether a new
    // thread needs to be started or not so synchronize on "this"

    synchronized (this) {
      final boolean isEmpty = isEmpty(); // Remember whether queue is currently empty or not

      // Add the invocation to the queue

      synchronized (barrier) {
        queue.add(invocation);

        queueSize += invocation.getSize();

        // Check to see if the queue limits have been exceeded and the queue should now be locked

        if (queueSize >= maxQueueSize || (maxQueueMsgs > 0 && queue.size() >= maxQueueMsgs)) {
          barrier.lock();
          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "Locked the barrier: bytes="+queueSize+" ("+maxQueueSize+") msgs="+queue.size()+" ("+maxQueueMsgs+")");
        } else {
          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "Leaving barrier unlocked: bytes="+queueSize+" ("+maxQueueSize+") msgs="+queue.size()+" ("+maxQueueMsgs+")");
        }
      }

      // If queue was previously empty we need to either prompt the existing processing thread or start a new thread to
      // process the new invocation request

      if (isEmpty) {
        if (running) { // A processing thread already exists for this queue so wake it up
          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "Notifying existing thread");
          notify();
        } else { // We need to start a new processing thread
          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "Starting a new thread");
          boolean interrupted = true;
          while (interrupted) {
            try {
              threadPool.execute(this);
              interrupted = false;
            } catch (InterruptedException e) {
              // No FFDC code needed
            }
          }
        }
      }
    } // synchronized (this)

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "enqueue");
  }
352035
private AbstractInvocation dequeue() {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "dequeue");

    AbstractInvocation invocation;

    synchronized (barrier) {
      invocation = queue.remove(0);
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "dequeue", invocation);
    return invocation;
  }
120163
public WorkgroupQueue getQueue(String queueName) {
        Resourcepart queueNameResourcepart;
        try {
            queueNameResourcepart = Resourcepart.from(queueName);
        }
        catch (XmppStringprepException e) {
            throw new IllegalArgumentException(e);
        }
        return getQueue(queueNameResourcepart);
    }
919371
public static <E> BlockingQueue<E> getLimitedBlockingQueue(int maxQueue) {
        return new LinkedBlockingQueue<E>(maxQueue) {
            @Override
            public boolean offer(E e) {
                return putInsteadOfOffer(this, e);
            }
        };
    }
1024868
private void executeAsyncCallRunnable(Runnable runnable, String callName, boolean blocking) {
		// make sure the executor is initialized. lock against concurrent calls to this function
		synchronized (this) {
			if (executionState != ExecutionState.RUNNING) {
				return;
			}

			// get ourselves a reference on the stack that cannot be concurrently modified
			BlockingCallMonitoringThreadPool executor = this.asyncCallDispatcher;
			if (executor == null) {
				// first time use, initialize
				checkState(userCodeClassLoader != null, "userCodeClassLoader must not be null");

				// Under normal execution, we expect that one thread will suffice, this is why we
				// keep the core threads to 1. In the case of a synchronous savepoint, we will block
				// the checkpointing thread, so we need an additional thread to execute the
				// notifyCheckpointComplete() callback. Finally, we aggressively purge (potentially)
				// idle thread so that we do not risk to have many idle thread on machines with multiple
				// tasks on them. Either way, only one of them can execute at a time due to the
				// checkpoint lock.

				executor = new BlockingCallMonitoringThreadPool(
						new DispatcherThreadFactory(
							TASK_THREADS_GROUP,
							"Async calls on " + taskNameWithSubtask,
							userCodeClassLoader));
				this.asyncCallDispatcher = executor;

				// double-check for execution state, and make sure we clean up after ourselves
				// if we created the dispatcher while the task was concurrently canceled
				if (executionState != ExecutionState.RUNNING) {
					executor.shutdown();
					asyncCallDispatcher = null;
					return;
				}
			}

			LOG.debug("Invoking async call {} on task {}", callName, taskNameWithSubtask);

			try {
				executor.submit(runnable, blocking);
			}
			catch (RejectedExecutionException e) {
				// may be that we are concurrently finished or canceled.
				// if not, report that something is fishy
				if (executionState == ExecutionState.RUNNING) {
					throw new RuntimeException("Async call with a " + (blocking ? "" : "non-") + "blocking call was rejected, even though the task is running.", e);
				}
			}
		}
	}