Java 类java.util.concurrent.SynchronousQueue 实例源码

private static ExecutorService newExecutorService(
    ThreadFactory threadFactory, String threadName) {
  boolean background = threadFactory instanceof GaeThreadFactory
      && ((GaeThreadFactory) threadFactory).isUsingBackgroundThreads();
  if (background) {
    // Create a thread pool with long-lived threads if background thread support is available.
    return new RevivingScheduledExecutor(threadFactory, threadName, true);
  } else {
    // Create an executor that creates a new thread for each submitted task, when background
    // thread support is not available.
    return new ThreadPoolExecutor(
        0,
        Integer.MAX_VALUE,
        0L,
        TimeUnit.SECONDS,
        new SynchronousQueue<Runnable>(),
        threadFactory);
  }
}

private static AsynchronousChannelGroup createAsynchronousChannelGroup() {
    // Need to do this with the right thread context class loader else the
    // first web app to call this will trigger a leak
    ClassLoader original = Thread.currentThread().getContextClassLoader();

    try {
        Thread.currentThread().setContextClassLoader(AsyncIOThreadFactory.class.getClassLoader());

        // These are the same settings as the default
        // AsynchronousChannelGroup
        int initialSize = Runtime.getRuntime().availableProcessors();
        ExecutorService executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, Long.MAX_VALUE,
                TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>(), new AsyncIOThreadFactory());

        try {
            return AsynchronousChannelGroup.withCachedThreadPool(executorService, initialSize);
        } catch (IOException e) {
            // No good reason for this to happen.
            throw new IllegalStateException(sm.getString("asyncChannelGroup.createFail"));
        }
    } finally {
        Thread.currentThread().setContextClassLoader(original);
    }
}

@Bean(name = "pool-server")
public TServer poolServer() throws Exception {
    TServerTransport transport = new TServerSocket(this.port());

    TThreadPoolServer.Args args = new TThreadPoolServer.Args(transport);
    args.transportFactory(new TTransportFactory());
    args.protocolFactory(new TBinaryProtocol.Factory());

    args.processor(this.processor());
    args.executorService(new ThreadPoolExecutor(env.getProperty(
            "rpc.server.min.worker.threads", Integer.class, 512), env
            .getProperty("rpc.server.max.worker.threads", Integer.class,
                    65535), env.getProperty(
            "rpc.server.thread.keep.alive.time", Long.class, 600l),
            TimeUnit.SECONDS, new SynchronousQueue<Runnable>()));

    return new TThreadPoolServer(args);
}

public synchronized void start(CommandExecutor commandExecutor) {
    if (isStarted()) {
        return;
    }

    this.commandExecutor = requireNonNull(commandExecutor, "commandExecutor");

    scheduler = MoreExecutors.listeningDecorator(Executors.newSingleThreadScheduledExecutor(
            new DefaultThreadFactory("mirroring-scheduler", true)));

    worker = MoreExecutors.listeningDecorator(
            new ThreadPoolExecutor(0, numThreads, 1, TimeUnit.MINUTES, new SynchronousQueue<>(),
                                   new DefaultThreadFactory("mirroring-worker", true)));

    final ListenableScheduledFuture<?> future = scheduler.scheduleWithFixedDelay(
            this::schedulePendingMirrors,
            TICK.getSeconds(), TICK.getSeconds(), TimeUnit.SECONDS);

    FuturesExtra.addFailureCallback(
            future,
            cause -> logger.error("Git-to-CD mirroring scheduler stopped due to an unexpected exception:",
                                  cause));
}

public void testGet_concurrent() {
  assertTrue(ArbitraryInstances.get(BlockingDeque.class).isEmpty());
  assertTrue(ArbitraryInstances.get(BlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(DelayQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(SynchronousQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(PriorityBlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentMap.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentNavigableMap.class).isEmpty());
  ArbitraryInstances.get(Executor.class).execute(ArbitraryInstances.get(Runnable.class));
  assertNotNull(ArbitraryInstances.get(ThreadFactory.class));
  assertFreshInstanceReturned(
      BlockingQueue.class, BlockingDeque.class, PriorityBlockingQueue.class,
      DelayQueue.class, SynchronousQueue.class,
      ConcurrentMap.class, ConcurrentNavigableMap.class,
      AtomicReference.class, AtomicBoolean.class,
      AtomicInteger.class, AtomicLong.class, AtomicDouble.class);
}

public FixedThreadPoolQueuesExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
        ThreadFactory threadFactory,WaitConditionStrategy waitConditionStrategy) {
    super(DEFAULT_INITIAL_THREAD_POOL_SIZE, 1, keepAliveTime, unit, new SynchronousQueue<Runnable>(),
            threadFactory, new AbortPolicy());

    if (corePoolSize < DEFAULT_INITIAL_THREAD_POOL_SIZE) {
        throw new IllegalArgumentException("corePoolSize: " + corePoolSize);
    }

    if ((maximumPoolSize == 0) || (maximumPoolSize < corePoolSize)) {
        throw new IllegalArgumentException("maximumPoolSize: " + maximumPoolSize);
    }
    if(waitConditionStrategy==null)
    {
         throw new IllegalArgumentException("waitConditionStrategy: " + waitConditionStrategy);
    }
    // Now, we can setup the pool sizes
    super.setCorePoolSize(corePoolSize);
    super.setMaximumPoolSize(maximumPoolSize);
    this.waitConditionStrategy=waitConditionStrategy;
}

/**
 * Creates a new instance of a OrderedThreadPoolExecutor.
 * 
 * @param corePoolSize The initial pool sizePoolSize
 * @param maximumPoolSize The maximum pool size
 * @param keepAliveTime Default duration for a thread
 * @param unit Time unit used for the keepAlive value
 * @param threadFactory The factory used to create threads
 * @param eventQueueHandler The queue used to store events
 */
public FixedThreadPoolBlockingQueuesExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
        ThreadFactory threadFactory) {
    // We have to initialize the pool with default values (0 and 1) in order to
    // handle the exception in a better way. We can't add a try {} catch() {}
    // around the super() call.
    super(DEFAULT_INITIAL_THREAD_POOL_SIZE, 1, keepAliveTime, unit, new SynchronousQueue<Runnable>(),
            threadFactory, new AbortPolicy());

    if (corePoolSize < DEFAULT_INITIAL_THREAD_POOL_SIZE) {
        throw new IllegalArgumentException("corePoolSize: " + corePoolSize);
    }

    if ((maximumPoolSize == 0) || (maximumPoolSize < corePoolSize)) {
        throw new IllegalArgumentException("maximumPoolSize: " + maximumPoolSize);
    }

    // Now, we can setup the pool sizes
    super.setCorePoolSize(corePoolSize);
    super.setMaximumPoolSize(maximumPoolSize);
}

/**
 * drainTo(c, n) empties up to n elements of queue into c
 */
public void testDrainToN() throws InterruptedException {
    final SynchronousQueue q = new SynchronousQueue();
    Thread t1 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            q.put(one);
        }});

    Thread t2 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            q.put(two);
        }});

    ArrayList l = new ArrayList();
    int drained;
    while ((drained = q.drainTo(l, 1)) == 0) Thread.yield();
    assertEquals(1, drained);
    assertEquals(1, l.size());
    while ((drained = q.drainTo(l, 1)) == 0) Thread.yield();
    assertEquals(1, drained);
    assertEquals(2, l.size());
    assertTrue(l.contains(one));
    assertTrue(l.contains(two));
    awaitTermination(t1);
    awaitTermination(t2);
}

public void testPollInExecutor(boolean fair) {
    final SynchronousQueue q = new SynchronousQueue(fair);
    final CheckedBarrier threadsStarted = new CheckedBarrier(2);
    final ExecutorService executor = Executors.newFixedThreadPool(2);
    try (PoolCleaner cleaner = cleaner(executor)) {
        executor.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                assertNull(q.poll());
                threadsStarted.await();
                assertSame(one, q.poll(LONG_DELAY_MS, MILLISECONDS));
                assertTrue(q.isEmpty());
            }});

        executor.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                threadsStarted.await();
                q.put(one);
            }});
    }
}

private static PooledExecutorWithDMStats createExecutor(PoolStatHelper poolHelper,
    final ThreadGroup threadGroup) {
  ThreadFactory factory = new ThreadFactory() {
    private final AtomicInteger threadNum = new AtomicInteger();

    public Thread newThread(Runnable r) {
      Thread thread = new Thread(threadGroup, r,
          "locator request thread[" + threadNum.incrementAndGet() + "]");
      thread.setDaemon(true);
      return thread;
    }
  };

  return new PooledExecutorWithDMStats(new SynchronousQueue(), MAX_POOL_SIZE, poolHelper, factory,
      POOL_IDLE_TIMEOUT, new ThreadPoolExecutor.CallerRunsPolicy());
}

public void testGet_concurrent() {
  assertTrue(ArbitraryInstances.get(BlockingDeque.class).isEmpty());
  assertTrue(ArbitraryInstances.get(BlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(DelayQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(SynchronousQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(PriorityBlockingQueue.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentMap.class).isEmpty());
  assertTrue(ArbitraryInstances.get(ConcurrentNavigableMap.class).isEmpty());
  ArbitraryInstances.get(Executor.class).execute(ArbitraryInstances.get(Runnable.class));
  assertNotNull(ArbitraryInstances.get(ThreadFactory.class));
  assertFreshInstanceReturned(
      BlockingQueue.class, BlockingDeque.class, PriorityBlockingQueue.class,
      DelayQueue.class, SynchronousQueue.class,
      ConcurrentMap.class, ConcurrentNavigableMap.class,
      AtomicReference.class, AtomicBoolean.class,
      AtomicInteger.class, AtomicLong.class, AtomicDouble.class);
}

/**
 * a deserialized/reserialized queue is usable
 */
public void testSerialization() {
    final SynchronousQueue x = new SynchronousQueue();
    final SynchronousQueue y = new SynchronousQueue(false);
    final SynchronousQueue z = new SynchronousQueue(true);
    assertSerialEquals(x, y);
    assertNotSerialEquals(x, z);
    SynchronousQueue[] qs = { x, y, z };
    for (SynchronousQueue q : qs) {
        SynchronousQueue clone = serialClone(q);
        assertNotSame(q, clone);
        assertSerialEquals(q, clone);
        assertTrue(clone.isEmpty());
        assertEquals(0, clone.size());
        assertEquals(0, clone.remainingCapacity());
        assertFalse(clone.offer(zero));
    }
}

public static void main(String[] args) throws Exception {
     final int maxConsumers = (args.length > 0)
         ? Integer.parseInt(args[0])
         : 5;

     pool = Executors.newCachedThreadPool();
     for (int i = 1; i <= maxConsumers; i += (i+1) >>> 1) {
         // Adjust iterations to limit typical single runs to <= 10 ms;
         // Notably, fair queues get fewer iters.
         // Unbounded queues can legitimately OOME if iterations
         // high enough, but we have a sufficiently low limit here.
         run(new ArrayBlockingQueue<Integer>(100), i, 1000);
         run(new LinkedBlockingQueue<Integer>(100), i, 1000);
         run(new LinkedBlockingDeque<Integer>(100), i, 1000);
         run(new LinkedTransferQueue<Integer>(), i, 700);
         run(new PriorityBlockingQueue<Integer>(), i, 1000);
         run(new SynchronousQueue<Integer>(), i, 300);
         run(new SynchronousQueue<Integer>(true), i, 200);
         run(new ArrayBlockingQueue<Integer>(100, true), i, 100);
     }
     pool.shutdown();
     if (! pool.awaitTermination(LONG_DELAY_MS, MILLISECONDS))
         throw new Error();
     pool = null;
}

@Before
public void setUp() throws Exception {
    this.count = 3;
    this.peers = new HashMap<Long,QuorumServer>(count);
    peerQuorumPort = new int[count];
    peerClientPort = new int[count];
    authzHosts = new HashSet<String>();
    for(int i = 0; i < count; i++) {
        peerQuorumPort[i] = PortAssignment.unique();
        peerClientPort[i] = PortAssignment.unique();
        QuorumServer qs = new QuorumServer(i, "0.0.0.0",
                peerQuorumPort[i], PortAssignment.unique(), null);
        peers.put(Long.valueOf(i), qs);
        authzHosts.add(qs.hostname);
    }
    executor = new ThreadPoolExecutor(3, 10,
            60, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
}

public void testDrainToWithActivePut(boolean fair) {
    final SynchronousQueue q = new SynchronousQueue(fair);
    Thread t = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            q.put(one);
        }});

    ArrayList l = new ArrayList();
    long startTime = System.nanoTime();
    while (l.isEmpty()) {
        q.drainTo(l);
        if (millisElapsedSince(startTime) > LONG_DELAY_MS)
            fail("timed out");
        Thread.yield();
    }
    assertEquals(1, l.size());
    assertSame(one, l.get(0));
    awaitTermination(t);
}

private BlockingQueue<Runnable> createBlockingQueue() {
    BlockingQueueTypeEnum queueType = BlockingQueueTypeEnum.fromString(txConfig.getBlockingQueueType());

    switch (queueType) {
        case LINKED_BLOCKING_QUEUE:
            return new LinkedBlockingQueue<>(1024);
        case ARRAY_BLOCKING_QUEUE:
            return new ArrayBlockingQueue<>(MAX_ARRAY_QUEUE);
        case SYNCHRONOUS_QUEUE:
            return new SynchronousQueue<>();
        default:
            return new LinkedBlockingQueue<>(1024);
    }

}

public static void main(String[] args)
{
    test("LinkedBlockingQueue", // Unlimited size
            new LinkedBlockingQueue<LiftOff>());
    test("ArrayBlockingQueue", // Fixed size
            new ArrayBlockingQueue<LiftOff>(3));
    test("SynchronousQueue", // Size of 1
            new SynchronousQueue<LiftOff>());
}

private BotContext() {
    fitnessQueue = new SynchronousQueue<>();
    individualsQueue = new SynchronousQueue<>();
    bot = new Bot(fitnessQueue, individualsQueue);
    workerThread = new Thread(bot);
    workerThread.start();
}

public void testListenInPoolThreadCustomExecutorInterrupted()
    throws Exception {
  final CountDownLatch submitSuccessful = new CountDownLatch(1);
  ExecutorService executorService = new ThreadPoolExecutor(
      0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
      new SynchronousQueue<Runnable>(),
      new ThreadFactoryBuilder().setDaemon(true).build()) {
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
      submitSuccessful.countDown();
    }
  };
  NonListenableSettableFuture<String> abstractFuture =
      NonListenableSettableFuture.create();
  ListenableFuture<String> listenableFuture =
      listenInPoolThread(abstractFuture, executorService);

  SingleCallListener singleCallListener = new SingleCallListener();
  singleCallListener.expectCall();

  assertFalse(singleCallListener.wasCalled());
  assertFalse(listenableFuture.isDone());

  listenableFuture.addListener(singleCallListener, directExecutor());
  /*
   * Don't shut down until the listenInPoolThread task has been accepted to
   * run. We want to see what happens when it's interrupted, not when it's
   * rejected.
   */
  submitSuccessful.await();
  executorService.shutdownNow();
  abstractFuture.set(DATA1);
  assertEquals(DATA1, listenableFuture.get());
  singleCallListener.waitForCall();

  assertTrue(singleCallListener.wasCalled());
  assertTrue(listenableFuture.isDone());
}

public RxWrapperDocumentClientImpl(DocumentClient client) {
    this.client = client;

    int maxThreads = (int) (client.getConnectionPolicy().getMaxPoolSize() * 1.1);
    this.executorService =  new ThreadPoolExecutor(
            Math.min(8, maxThreads), // core thread pool size
            maxThreads, // maximum thread pool size
            30, // time to wait before killing idle threads
            TimeUnit.SECONDS, 
            new SynchronousQueue<>(),
            new RxThreadFactory("RxDocdb-io"),
            new ThreadPoolExecutor.CallerRunsPolicy());
    this.scheduler = Schedulers.from(executorService);
}

private String selectAccount(Account accounts[]) {
  final SynchronousQueue<String> queue = new SynchronousQueue<String>();
  SelectAccount select = new SelectAccount(accounts, queue);
  select.start();
  Log.i(LOG_TAG, "Select: waiting for user...");
  String account = null;
  try {
    account = queue.take();
  } catch (InterruptedException e) {
    e.printStackTrace();
  }
  Log.i(LOG_TAG, "Selected: " + account);
  return account == NO_ACCOUNT ? null : account;
}

public void testIteratorRemove(boolean fair) {
    final SynchronousQueue q = new SynchronousQueue(fair);
    Iterator it = q.iterator();
    try {
        it.remove();
        shouldThrow();
    } catch (IllegalStateException success) {}
}

public static ExecutorService newCachedThreadPool(ThreadFactory
    threadFactory) {
  return new HadoopThreadPoolExecutor(0, Integer.MAX_VALUE,
      60L, TimeUnit.SECONDS,
      new SynchronousQueue<Runnable>(),
      threadFactory);
}

void test(String[] args) throws Throwable {
    final int n = 100;
    final ThreadPoolExecutor pool =
        new ThreadPoolExecutor(n, n, 1L, TimeUnit.NANOSECONDS,
                               new SynchronousQueue<Runnable>());
    final CountDownLatch startingGate = new CountDownLatch(n);
    final CountDownLatch finishLine = new CountDownLatch(n);
    equal(pool.getCorePoolSize(), n);
    equal(pool.getPoolSize(), 0);
    for (int i = 0; i < n; i++)
        pool.execute(new Runnable() { public void run() {
            try {
                startingGate.countDown();
                startingGate.await();
                equal(pool.getPoolSize(), n);
                pool.setCorePoolSize(n);
                pool.setCorePoolSize(1);
                check(! Thread.interrupted());
                equal(pool.getPoolSize(), n);
                finishLine.countDown();
                finishLine.await();
                check(! Thread.interrupted());
            } catch (Throwable t) { unexpected(t); }}});
    finishLine.await();
    pool.shutdown();
    check(pool.awaitTermination(LONG_DELAY_MS, MILLISECONDS));
}

private PCloudAPIClient(Builder builder) {
    this.connectTimeoutMs = builder.connectTimeoutMs;
    this.writeTimeoutMs = builder.writeTimeoutMs;
    this.readTimeoutMs = builder.readTimeoutMs;

    this.socketFactory = builder.socketFactory != null ? builder.socketFactory : SocketFactory.getDefault();

    this.sslSocketFactory = builder.sslSocketFactory != null ?
                    builder.sslSocketFactory : (SSLSocketFactory) SSLSocketFactory.getDefault();

    this.hostnameVerifier = builder.hostnameVerifier != null ?
                    builder.hostnameVerifier : DefaultHostnameVerifier.INSTANCE;

    this.connectionPool = builder.connectionPool != null ? builder.connectionPool : new ConnectionPool();
    this.endpointProvider = builder.endpointProvider != null ? builder.endpointProvider : EndpointProvider.DEFAULT;

    ConnectionFactory connectionFactory = new ConnectionFactory(socketFactory, sslSocketFactory, hostnameVerifier);
    this.connectionProvider = new ConnectionProvider(connectionPool, endpointProvider, connectionFactory,
            connectTimeoutMs, readTimeoutMs, writeTimeoutMs, false);

    ThreadFactory threadFactory = new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "PCloud API Client");
        }
    };
    this.callExecutor = builder.callExecutor != null ?
            builder.callExecutor : new ThreadPoolExecutor(0,
            Integer.MAX_VALUE,
            DEFAULT_KEEP_ALIVE_TIME_MS,
            TimeUnit.SECONDS,
            new SynchronousQueue<Runnable>(),
            threadFactory);

    this.interceptors = Collections.unmodifiableList(new ArrayList<>(builder.interceptors));
}

public Executor getExecutor(URL url) {
    String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
    int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
    int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
    return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS, 
            queues == 0 ? new SynchronousQueue<Runnable>() : 
                (queues < 0 ? new LinkedBlockingQueue<Runnable>() 
                        : new LinkedBlockingQueue<Runnable>(queues)),
            new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
}

public Executor getExecutor(URL url) {
    String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
    int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
    int threads = url.getParameter(Constants.THREADS_KEY, Integer.MAX_VALUE);
    int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
    int alive = url.getParameter(Constants.ALIVE_KEY, Constants.DEFAULT_ALIVE);
    return new ThreadPoolExecutor(cores, threads, alive, TimeUnit.MILLISECONDS, 
            queues == 0 ? new SynchronousQueue<Runnable>() : 
                (queues < 0 ? new LinkedBlockingQueue<Runnable>() 
                        : new LinkedBlockingQueue<Runnable>(queues)),
            new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
}

public Executor getExecutor(URL url) {
    String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
    int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
    int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
    int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
    return new ThreadPoolExecutor(cores, threads, Long.MAX_VALUE, TimeUnit.MILLISECONDS, 
            queues == 0 ? new SynchronousQueue<Runnable>() : 
                (queues < 0 ? new LinkedBlockingQueue<Runnable>() 
                        : new LinkedBlockingQueue<Runnable>(queues)),
            new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
}

/**
 * Get a pool that has only ever one thread. A second action added to the pool (running
 * concurrently), will cause an exception.
 * @return
 */
private ExecutorService getPool() {
  int maxThreads = 1;
  long keepAliveTime = 60;
  ThreadPoolExecutor pool =
      new ThreadPoolExecutor(1, maxThreads, keepAliveTime, TimeUnit.SECONDS,
          new SynchronousQueue<Runnable>(), Threads.newDaemonThreadFactory("hbase-table"));
  pool.allowCoreThreadTimeOut(true);
  return pool;
}

public Executor getExecutor(URL url) {
    String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
    int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
    int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
    int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
    return new ThreadPoolExecutor(cores, threads, Long.MAX_VALUE, TimeUnit.MILLISECONDS, 
            queues == 0 ? new SynchronousQueue<Runnable>() : 
                (queues < 0 ? new LinkedBlockingQueue<Runnable>() 
                        : new LinkedBlockingQueue<Runnable>(queues)),
            new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
}

public LifecycleModule() {
    // The thread pool is unbounded, so use direct handoff
    BlockingQueue<Runnable> queue = new SynchronousQueue<Runnable>();
    // Discard tasks that are submitted during shutdown
    RejectedExecutionHandler policy =
            new ThreadPoolExecutor.DiscardPolicy();
    // Create threads as required and keep them in the pool for 60 seconds
    ioExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
            60, SECONDS, queue, policy);
}

private static AsynchronousChannelGroup createAsynchronousChannelGroup() {
    // Need to do this with the right thread context class loader else the
    // first web app to call this will trigger a leak
    ClassLoader original = Thread.currentThread().getContextClassLoader();

    try {
        Thread.currentThread().setContextClassLoader(
                AsyncIOThreadFactory.class.getClassLoader());

        // These are the same settings as the default
        // AsynchronousChannelGroup
        int initialSize = Runtime.getRuntime().availableProcessors();
        ExecutorService executorService = new ThreadPoolExecutor(
                0,
                Integer.MAX_VALUE,
                Long.MAX_VALUE, TimeUnit.MILLISECONDS,
                new SynchronousQueue<Runnable>(),
                new AsyncIOThreadFactory());

        try {
            return AsynchronousChannelGroup.withCachedThreadPool(
                    executorService, initialSize);
        } catch (IOException e) {
            // No good reason for this to happen.
            throw new IllegalStateException(sm.getString("asyncChannelGroup.createFail"));
        }
    } finally {
        Thread.currentThread().setContextClassLoader(original);
    }
}

public Executor getExecutor(URL url) {
    String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
    int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
    int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
    return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS, 
            queues == 0 ? new SynchronousQueue<Runnable>() : 
                (queues < 0 ? new LinkedBlockingQueue<Runnable>() 
                        : new LinkedBlockingQueue<Runnable>(queues)),
            new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
}

public Executor getExecutor(URL url) {
    String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
    int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
    int threads = url.getParameter(Constants.THREADS_KEY, Integer.MAX_VALUE);
    int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
    int alive = url.getParameter(Constants.ALIVE_KEY, Constants.DEFAULT_ALIVE);
    return new ThreadPoolExecutor(cores, threads, alive, TimeUnit.MILLISECONDS, 
            queues == 0 ? new SynchronousQueue<Runnable>() : 
                (queues < 0 ? new LinkedBlockingQueue<Runnable>() 
                        : new LinkedBlockingQueue<Runnable>(queues)),
            new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
}

public static List<BlockingQueue<Object>> blockingQueues() {
  return ImmutableList.<BlockingQueue<Object>>of(
      new LinkedBlockingQueue<Object>(),
      new LinkedBlockingQueue<Object>(10),
      new SynchronousQueue<Object>(),
      new ArrayBlockingQueue<Object>(10),
      new LinkedBlockingDeque<Object>(),
      new LinkedBlockingDeque<Object>(10),
      new PriorityBlockingQueue<Object>(10, Ordering.arbitrary()));
}

private void initializeAuth(final long mySid,
        final QuorumAuthServer authServer,
        final QuorumAuthLearner authLearner,
        final int quorumCnxnThreadsSize,
        final boolean quorumSaslAuthEnabled) {
    this.authServer = authServer;
    this.authLearner = authLearner;
    this.quorumSaslAuthEnabled = quorumSaslAuthEnabled;
    if (!this.quorumSaslAuthEnabled) {
        LOG.debug("Not initializing connection executor as quorum sasl auth is disabled");
        return;
    }

    // init connection executors
    final AtomicInteger threadIndex = new AtomicInteger(1);
    SecurityManager s = System.getSecurityManager();
    final ThreadGroup group = (s != null) ? s.getThreadGroup()
            : Thread.currentThread().getThreadGroup();
    ThreadFactory daemonThFactory = new ThreadFactory() {

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, "QuorumConnectionThread-"
                    + "[myid=" + mySid + "]-"
                    + threadIndex.getAndIncrement());
            return t;
        }
    };
    this.connectionExecutor = new ThreadPoolExecutor(3,
            quorumCnxnThreadsSize, 60, TimeUnit.SECONDS,
            new SynchronousQueue<Runnable>(), daemonThFactory);
    this.connectionExecutor.allowCoreThreadTimeOut(true);
}

/**
 * Construct a ConcurrentBag with the specified listener.
 *
 * @param listener the IBagStateListener to attach to this bag
 */
public ConcurrentBag(IBagStateListener listener) {
    this.listener = listener;
    this.weakThreadLocals = useWeakThreadLocals();

    this.handoffQueue = new SynchronousQueue<>(true);
    this.waiters = new AtomicInteger();
    this.sharedList = new CopyOnWriteArrayList<>();
    if (weakThreadLocals) {
        this.threadList = ThreadLocal.withInitial(() -> new ArrayList<>(16));
    } else {
        this.threadList = ThreadLocal.withInitial(() ->
                           new FastList<>(IConcurrentBagEntry.class, 16));
    }
}

public static SynchronousQueue<Object> getMessageQueue(String channel) {
  SynchronousQueue<Object> result = messagePool.get(channel);
  if (result == null) {
    result = new SynchronousQueue<Object>();
    messagePool.put(channel, result);
  }
  return result;
}

public static RealTimeThreadPool getInstance() {
    if (null == mInstance) {
        synchronized (lock) {
            if (null == mInstance) {
                mInstance = new RealTimeThreadPool();
                executorService = new ThreadPoolExecutor(1, 10, 120, TimeUnit.SECONDS, new
                        SynchronousQueue<Runnable>(),
                        new DefaultThreadFactory(), new ThreadPoolExecutor.DiscardPolicy());
            }
        }
    }
    return mInstance;
}