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

@Override
protected void doGet(HttpServletRequest req, HttpServletResponse resp)
        throws ServletException, IOException {

    resp.getWriter().println(
            "The current thread served " + this + " servlet");
    tpe = new ThreadPoolExecutor(tpSize, tpSize, 50000L,
            TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());

    Task[] tasks = new Task[nTasks];
    for (int i = 0; i < nTasks; i++) {
        tasks[i] = new Task("Task " + i);
        tpe.execute(tasks[i]);
    }
    resp.getWriter().println("Started " + nTasks +
            " never ending tasks using the ThreadPoolExecutor");
    resp.getWriter().flush();
}

/**
 * submit(callable).get() throws InterruptedException if interrupted
 */
public void testInterruptedSubmit() throws InterruptedException {
    final CountDownLatch submitted    = new CountDownLatch(1);
    final CountDownLatch quittingTime = new CountDownLatch(1);
    final Callable<Void> awaiter = new CheckedCallable<Void>() {
        public Void realCall() throws InterruptedException {
            assertTrue(quittingTime.await(2*LONG_DELAY_MS, MILLISECONDS));
            return null;
        }};
    final ExecutorService p
        = new ThreadPoolExecutor(1,1,60, TimeUnit.SECONDS,
                                 new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(p, quittingTime)) {
        Thread t = newStartedThread(new CheckedInterruptedRunnable() {
            public void realRun() throws Exception {
                Future<Void> future = p.submit(awaiter);
                submitted.countDown();
                future.get();
            }});

        await(submitted);
        t.interrupt();
        awaitTermination(t);
    }
}

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 ThreadPoolExecutor createExecutor() {
    return new ThreadPoolExecutor(
        Config.concurrentThreadNum(),
        Config.concurrentThreadNum(),
        0L,
        TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(),
        defaultFactory,
        (r, executor) -> {
            if (r instanceof TaskRunner) {
                TaskRunner task = (TaskRunner) r;
                onReject(task.getCmd());
                LOGGER.warn("Reject cmd: %s", task.getCmd());
            }
        });
}

public CsapEventClient( ) {

        BasicThreadFactory eventThreadFactory = new BasicThreadFactory.Builder()
            .namingPattern( "CsapEventPost-%d" )
            .daemon( true )
            .priority( Thread.NORM_PRIORITY + 1 )
            .build();

        eventPostQueue = new ArrayBlockingQueue<>( MAX_EVENT_BACKLOG );
        // Use a single thread to sequence and post
        // eventPostPool = Executors.newFixedThreadPool(1, schedFactory, queue);
        // really only needs to be 1 - adding the others for lt scenario
        eventPostPool = new ThreadPoolExecutor( 1, 1,
            30, TimeUnit.SECONDS,
            eventPostQueue, eventThreadFactory );

        eventPostCompletionService = new ExecutorCompletionService<String>(
            eventPostPool );
    }

protected String getThreadPoolSummary(ThreadPool threadPool) {
    StringBuffer sb = new StringBuffer(512);
    sb.append(threadPool.getName());

    if (threadPool.getExecutor() instanceof ThreadPoolExecutor) {
        sb.append(" (");
        ThreadPoolExecutor executor = (ThreadPoolExecutor) threadPool.getExecutor();
        sb.append("max=" + executor.getMaximumPoolSize());
        sb.append(", current=" + executor.getPoolSize());
        sb.append(", active=" + executor.getActiveCount());
        sb.append(", largest=" + executor.getLargestPoolSize());
        sb.append(", core=" + executor.getCorePoolSize());
        sb.append(", all tasks=" + executor.getTaskCount());
        sb.append(", completed tasks=" + executor.getCompletedTaskCount());
        sb.append(", queue size=" + executor.getQueue().size());
        sb.append(", queue remaining capacity=" + executor.getQueue().remainingCapacity());
        sb.append(")");
    }

    return sb.toString();
}

/**
 * invokeAll(c) throws NPE if c has null elements
 */
public void testInvokeAll3() throws Exception {
    final ExecutorService e =
        new ThreadPoolExecutor(2, 2,
                               LONG_DELAY_MS, MILLISECONDS,
                               new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(e)) {
        List<Callable<String>> l = new ArrayList<>();
        l.add(new StringTask());
        l.add(null);
        try {
            e.invokeAll(l);
            shouldThrow();
        } catch (NullPointerException success) {}
    }
}

/**
 * getActiveCount increases but doesn't overestimate, when a
 * thread becomes active
 */
public void testGetActiveCount() throws InterruptedException {
    final CountDownLatch done = new CountDownLatch(1);
    final ThreadPoolExecutor p =
        new ThreadPoolExecutor(2, 2,
                               LONG_DELAY_MS, MILLISECONDS,
                               new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(p, done)) {
        final CountDownLatch threadStarted = new CountDownLatch(1);
        assertEquals(0, p.getActiveCount());
        p.execute(new CheckedRunnable() {
            public void realRun() throws InterruptedException {
                threadStarted.countDown();
                assertEquals(1, p.getActiveCount());
                await(done);
            }});
        await(threadStarted);
        assertEquals(1, p.getActiveCount());
    }
}

private void initDictionaryCaches(DictionaryDAOImpl dictionaryDAO, TenantService tenantService)
{
    CompiledModelsCache compiledModelsCache = new CompiledModelsCache();
    compiledModelsCache.setDictionaryDAO(dictionaryDAO);
    compiledModelsCache.setTenantService(tenantService);
    compiledModelsCache.setRegistry(new DefaultAsynchronouslyRefreshedCacheRegistry());
    TraceableThreadFactory threadFactory = new TraceableThreadFactory();
    threadFactory.setThreadDaemon(true);
    threadFactory.setThreadPriority(Thread.NORM_PRIORITY);

    ThreadPoolExecutor threadPoolExecutor = new DynamicallySizedThreadPoolExecutor(20, 20, 90, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), threadFactory,
            new ThreadPoolExecutor.CallerRunsPolicy());
    compiledModelsCache.setThreadPoolExecutor(threadPoolExecutor);
    dictionaryDAO.setDictionaryRegistryCache(compiledModelsCache);
    dictionaryDAO.init();
}

/**
 * This is for testing the active number of threads that were used while
 * doing a batch operation. It inserts one row per region via the batch
 * operation, and then checks the number of active threads.
 * For HBASE-3553
 * @throws IOException
 * @throws InterruptedException
 * @throws NoSuchFieldException
 * @throws SecurityException
 */
@Ignore ("Nice bug flakey... expected 5 but was 4..") @Test(timeout=300000)
public void testActiveThreadsCount() throws Exception {
  try (Connection connection = ConnectionFactory.createConnection(UTIL.getConfiguration())) {
    ThreadPoolExecutor executor = HTable.getDefaultExecutor(UTIL.getConfiguration());
    try {
      try (Table t = connection.getTable(TEST_TABLE, executor)) {
        List<Put> puts = constructPutRequests(); // creates a Put for every region
        t.batch(puts);
        HashSet<ServerName> regionservers = new HashSet<ServerName>();
        try (RegionLocator locator = connection.getRegionLocator(TEST_TABLE)) {
          for (Row r : puts) {
            HRegionLocation location = locator.getRegionLocation(r.getRow());
            regionservers.add(location.getServerName());
          }
        }
        assertEquals(regionservers.size(), executor.getLargestPoolSize());
      }
    } finally {
      executor.shutdownNow();
    }
  }
}

/**
 * prestartCoreThread starts a thread if under corePoolSize, else doesn't
 */
public void testPrestartCoreThread() {
    final ThreadPoolExecutor p =
        new ThreadPoolExecutor(2, 6,
                               LONG_DELAY_MS, MILLISECONDS,
                               new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(p)) {
        assertEquals(0, p.getPoolSize());
        assertTrue(p.prestartCoreThread());
        assertEquals(1, p.getPoolSize());
        assertTrue(p.prestartCoreThread());
        assertEquals(2, p.getPoolSize());
        assertFalse(p.prestartCoreThread());
        assertEquals(2, p.getPoolSize());
        p.setCorePoolSize(4);
        assertTrue(p.prestartCoreThread());
        assertEquals(3, p.getPoolSize());
        assertTrue(p.prestartCoreThread());
        assertEquals(4, p.getPoolSize());
        assertFalse(p.prestartCoreThread());
        assertEquals(4, p.getPoolSize());
    }
}

/**
 * get of element of invokeAll(c) throws exception on failed task
 */
public void testInvokeAll4() throws Exception {
    final ExecutorService e =
        new ThreadPoolExecutor(2, 2,
                               LONG_DELAY_MS, MILLISECONDS,
                               new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(e)) {
        List<Callable<String>> l = new ArrayList<>();
        l.add(new NPETask());
        List<Future<String>> futures = e.invokeAll(l);
        assertEquals(1, futures.size());
        try {
            futures.get(0).get();
            shouldThrow();
        } catch (ExecutionException success) {
            assertTrue(success.getCause() instanceof NullPointerException);
        }
    }
}

@Override
protected void serviceInit(Configuration conf) throws Exception {
  int threadCount = conf.getInt(
      YarnConfiguration.RM_AMLAUNCHER_THREAD_COUNT,
      YarnConfiguration.DEFAULT_RM_AMLAUNCHER_THREAD_COUNT);
  ThreadFactory tf = new ThreadFactoryBuilder()
      .setNameFormat("ApplicationMasterLauncher #%d")
      .build();
  launcherPool = new ThreadPoolExecutor(threadCount, threadCount, 1,
      TimeUnit.HOURS, new LinkedBlockingQueue<Runnable>());
  launcherPool.setThreadFactory(tf);

  Configuration newConf = new YarnConfiguration(conf);
  newConf.setInt(CommonConfigurationKeysPublic.
          IPC_CLIENT_CONNECT_MAX_RETRIES_ON_SOCKET_TIMEOUTS_KEY,
      conf.getInt(YarnConfiguration.RM_NODEMANAGER_CONNECT_RETIRES,
          YarnConfiguration.DEFAULT_RM_NODEMANAGER_CONNECT_RETIRES));
  setConfig(newConf);
  super.serviceInit(newConf);
}

private ThreadPoolExecutor multiThreadUpload(int threadNum, final int threadFileNum) {

    ThreadPoolExecutor pool = (ThreadPoolExecutor) Executors.newFixedThreadPool(threadNum);
    pool.prestartAllCoreThreads();

    for (int i = 0; i < threadNum; ++i) {
      final int threadId = i;
      pool.submit(new Runnable() {
        @Override
        public void run() {
          uploadAndDownloadPerform(threadId, threadFileNum);
        }
      });
    }
    pool.shutdown();
    return pool;
  }

@Setup(Level.Trial)
@Override
public void setUp() throws Exception {
    ListeningExecutorService dsExec = MoreExecutors.newDirectExecutorService();
    executor = MoreExecutors.listeningDecorator(
            MoreExecutors.getExitingExecutorService((ThreadPoolExecutor) Executors.newFixedThreadPool(1), 1L,
                    TimeUnit.SECONDS));

    InMemoryDOMDataStore operStore = new InMemoryDOMDataStore("OPER", dsExec);
    InMemoryDOMDataStore configStore = new InMemoryDOMDataStore("CFG", dsExec);
    Map<LogicalDatastoreType, DOMStore> datastores = ImmutableMap.of(
        LogicalDatastoreType.OPERATIONAL, (DOMStore)operStore,
        LogicalDatastoreType.CONFIGURATION, configStore);

    domBroker = new SerializedDOMDataBroker(datastores, executor);
    schemaContext = BenchmarkModel.createTestContext();
    configStore.onGlobalContextUpdated(schemaContext);
    operStore.onGlobalContextUpdated(schemaContext);
    initTestNode();
}

@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    String msg = String.format("Thread pool is EXHAUSTED!" +
            " Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: %d)," +
            " Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!" ,
            threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(), e.getLargestPoolSize(),
            e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),
            url.getProtocol(), url.getIp(), url.getPort());
    logger.warn(msg);
    throw new RejectedExecutionException(msg);
}

/**
 * Constructor throws if keepAliveTime is less than zero
 */
public void testConstructor9() {
    try {
        new ThreadPoolExecutor(1, 2, -1L, SECONDS,
                               new ArrayBlockingQueue<Runnable>(10),
                               new SimpleThreadFactory());
        shouldThrow();
    } catch (IllegalArgumentException success) {}
}

/**
 * A new SingleThreadExecutor cannot be casted to concrete implementation
 */
public void testCastNewSingleThreadExecutor() {
    final ExecutorService e = Executors.newSingleThreadExecutor();
    try (PoolCleaner cleaner = cleaner(e)) {
        try {
            ThreadPoolExecutor tpe = (ThreadPoolExecutor)e;
            shouldThrow();
        } catch (ClassCastException success) {}
    }
}

/**
 * 固定大小线程池，无队列
 *
 * @param corePoolSize
 *         初始化线程池
 * @return the thread pool executor
 */
public static ThreadPoolExecutor newFixedThreadPool(int corePoolSize) {
    return new ThreadPoolExecutor(corePoolSize,
            corePoolSize,
            0,
            TimeUnit.MILLISECONDS,
            new SynchronousQueue<Runnable>());
}

/**
 * setCorePoolSize of negative value throws IllegalArgumentException
 */
public void testCorePoolSizeIllegalArgumentException() {
    final ThreadPoolExecutor p =
        new CustomTPE(1, 2,
                      LONG_DELAY_MS, MILLISECONDS,
                      new ArrayBlockingQueue<Runnable>(10));
    try (PoolCleaner cleaner = cleaner(p)) {
        try {
            p.setCorePoolSize(-1);
            shouldThrow();
        } catch (IllegalArgumentException success) {}
    }
}

public XExecutor getExecutor() {
    if (executor == null) {
        synchronized (UploadThreadPool.class) {
            if (executor == null) {
                executor = new XExecutor(corePoolSize, MAX_IMUM_POOL_SIZE, KEEP_ALIVE_TIME, UNIT, //
                                         new PriorityBlockingQueue<Runnable>(),   //无限容量的缓冲队列
                                         Executors.defaultThreadFactory(),        //线程创建工厂
                                         new ThreadPoolExecutor.AbortPolicy());   //继续超出上限的策略，阻止
            }
        }
    }
    return executor;
}

@Override
public void rejectedExecution(Runnable runnable, ThreadPoolExecutor executor) {
    if (threadName != null) {
        LOG.error("txTransaction Thread pool [{}] is exhausted, executor={}", threadName, executor.toString());
    }

    if (!executor.isShutdown()) {
        BlockingQueue<Runnable> queue = executor.getQueue();
        int discardSize = queue.size() >> 1;
        for (int i = 0; i < discardSize; i++) {
            queue.poll();
        }
        queue.offer(runnable);
    }
}

public static ExecutorService newDynamicSingleThreadedExecutor() {
  ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECONDS,
                                                       new LinkedBlockingQueue<Runnable>());
  executor.allowCoreThreadTimeOut(true);

  return executor;
}

public static Queue<FileParcel> convertFile(String[] paths) throws Exception {

        if (paths == null || paths.length == 0)
            return null;
        int tmp = Runtime.getRuntime().availableProcessors() + 1;
        int result = paths.length > tmp ? tmp : paths.length;
        Executor mExecutor = Executors.newFixedThreadPool(result);
        final Queue<FileParcel> mQueue = new LinkedBlockingQueue<>();
        CountDownLatch mCountDownLatch = new CountDownLatch(paths.length);

        int i = 1;
        for (String path : paths) {

            LogUtils.i("Info", "path   :  :" + path);
            if (TextUtils.isEmpty(path)) {
                mCountDownLatch.countDown();
                continue;
            }

            mExecutor.execute(new EncodeFileRunnable(path, mQueue, mCountDownLatch, i++));

        }
        mCountDownLatch.await();

        if (!((ThreadPoolExecutor) mExecutor).isShutdown())
            ((ThreadPoolExecutor) mExecutor).shutdownNow();

        LogUtils.i("Info", "isShutDown:" + (((ThreadPoolExecutor) mExecutor).isShutdown()));
        return mQueue;
    }

/**
 * Returns a new thread pool configured with the default settings.
 *
 * @return A new thread pool configured with the default settings.
 */
public static ThreadPoolExecutor createDefaultExecutorService() {
    ThreadFactory threadFactory = new ThreadFactory() {
        private int threadCount = 1;

        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("s3-transfer-manager-worker-" + threadCount++);
            return thread;
        }
    };
    return (ThreadPoolExecutor)Executors.newFixedThreadPool(10, threadFactory);
}

public void rejectedExecution(Runnable task, ThreadPoolExecutor executor)
{
    if (executor.isShutdown())
    {
        //Give some notification to the caller the task isn't going to run
        if (task instanceof Future)
            ((Future) task).cancel(false);

        logger.debug("ScheduledThreadPoolExecutor has shut down as part of C* shutdown");
    }
    else
    {
        throw new AssertionError("Unknown rejection of ScheduledThreadPoolExecutor task");
    }
}

public AsyncPool(Server server, int size) {
    this.currentThread = new AtomicInteger();
    this.size = size;
    this.pool = new ThreadPoolExecutor(size, Integer.MAX_VALUE,
            60, TimeUnit.MILLISECONDS, new SynchronousQueue<>(),
            runnable -> new Thread(runnable) {{
                setDaemon(true);
                setName(String.format("Nukkit Asynchronous Task Handler #%s", currentThread.incrementAndGet()));
            }}
    );
    this.server = server;
}

void test(String[] args) throws Throwable {
    final int threadCount = 10;
    final int timeoutMillis = 30;
    BlockingQueue<Runnable> q = new ArrayBlockingQueue<>(2*threadCount);
    ThreadPoolExecutor tpe
        = new ThreadPoolExecutor(threadCount, threadCount,
                                 timeoutMillis, TimeUnit.MILLISECONDS,
                                 q, new IdentifiableThreadFactory());
    equal(tpe.getCorePoolSize(), threadCount);
    check(! tpe.allowsCoreThreadTimeOut());
    tpe.allowCoreThreadTimeOut(true);
    check(tpe.allowsCoreThreadTimeOut());
    equal(countExecutorThreads(), 0);
    long startTime = System.nanoTime();
    for (int i = 0; i < threadCount; i++) {
        tpe.submit(() -> {});
        int count = countExecutorThreads();
        if (millisElapsedSince(startTime) < timeoutMillis)
            equal(count, i + 1);
    }
    while (countExecutorThreads() > 0 &&
           millisElapsedSince(startTime) < LONG_DELAY_MS)
        Thread.yield();
    equal(countExecutorThreads(), 0);
    check(millisElapsedSince(startTime) >= timeoutMillis);
    tpe.shutdown();
    check(tpe.allowsCoreThreadTimeOut());
    check(tpe.awaitTermination(LONG_DELAY_MS, MILLISECONDS));

    System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed);
    if (failed > 0) throw new Exception("Some tests failed");
}

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));
}

HttpServer(
        final int port,
        final InetAddress ifAddress,
        final SocketConfig socketConfig,
        final ServerSocketFactory serverSocketFactory,
        final HttpService httpService,
        final HttpConnectionFactory<? extends DefaultBHttpServerConnection> connectionFactory,
        final SSLServerSetupHandler sslSetupHandler,
        final ExceptionLogger exceptionLogger) {
    this.port = port;
    this.ifAddress = ifAddress;
    this.socketConfig = socketConfig;
    this.serverSocketFactory = serverSocketFactory;
    this.httpService = httpService;
    this.connectionFactory = connectionFactory;
    this.sslSetupHandler = sslSetupHandler;
    this.exceptionLogger = exceptionLogger;
    this.listenerExecutorService = new ThreadPoolExecutor(
            1, 1, 0L, TimeUnit.MILLISECONDS,
            new SynchronousQueue<Runnable>(),
            new ThreadFactoryImpl("HTTP-listener-" + this.port));
    this.workerThreads = new ThreadGroup("HTTP-workers");
    this.workerExecutorService = new WorkerPoolExecutor(
            0, Integer.MAX_VALUE, 1L, TimeUnit.SECONDS,
            new SynchronousQueue<Runnable>(),
            new ThreadFactoryImpl("HTTP-worker", this.workerThreads));
    this.status = new AtomicReference<Status>(Status.READY);
}

public void testGetExitingExcutorService_shutdownHookRegistered() throws InterruptedException {
  TestApplication application = new TestApplication();
  ThreadPoolExecutor executor = mock(ThreadPoolExecutor.class);
  ThreadFactory threadFactory = mock(ThreadFactory.class);
  when(executor.getThreadFactory()).thenReturn(threadFactory);
  ExecutorService unused = application.getExitingExecutorService(executor);
  application.shutdown();
  verify(executor).shutdown();
}

/**
 * Returns a new fixed thread pool with the given thread count, thread name prefix,
 * and {@link com.bumptech.glide.load.engine.executor.GlideExecutor.UncaughtThrowableStrategy}.
 *
 * <p>Source executors allow network operations on their threads.
 *
 * @param threadCount The number of threads.
 * @param name The prefix for each thread name.
 * @param uncaughtThrowableStrategy The {@link
 * com.bumptech.glide.load.engine.executor.GlideExecutor.UncaughtThrowableStrategy} to use to
 *                                  handle uncaught exceptions.
 */
// Public API.
@SuppressWarnings("WeakerAccess")
public static GlideExecutor newSourceExecutor(
    int threadCount, String name, UncaughtThrowableStrategy uncaughtThrowableStrategy) {
  return new GlideExecutor(
      new ThreadPoolExecutor(
          threadCount /* corePoolSize */,
          threadCount /* maximumPoolSize */,
          0 /* keepAliveTime */,
          TimeUnit.MILLISECONDS,
          new PriorityBlockingQueue<Runnable>(),
          new DefaultThreadFactory(name, uncaughtThrowableStrategy, false)));
}

private ThreadPoolExecutor getThreadPoolExecutor ()
   throws IllegalStateException
{
   Assert.state (this.threadPoolExecutor != null,
      "ThreadPoolTaskExecutor not initialized");
   return this.threadPoolExecutor;
}

public Status check() {
    DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
    Map<String, Object> executors = dataStore.get(Constants.EXECUTOR_SERVICE_COMPONENT_KEY);

    StringBuilder msg = new StringBuilder();
    Status.Level level = Status.Level.OK;
    for(Map.Entry<String, Object> entry : executors.entrySet()) {
        String port = entry.getKey();
        ExecutorService executor = (ExecutorService) entry.getValue();

        if (executor != null && executor instanceof ThreadPoolExecutor) {
            ThreadPoolExecutor tp = (ThreadPoolExecutor) executor;
            boolean ok = tp.getActiveCount() < tp.getMaximumPoolSize() - 1;
            Status.Level lvl = Status.Level.OK;
            if(!ok) {
                level = Status.Level.WARN;
                lvl = Status.Level.WARN;
            }

            if(msg.length() > 0) {
                msg.append(";");
            }
            msg.append("Pool status:" + lvl
                    + ", max:" + tp.getMaximumPoolSize()
                    + ", core:" + tp.getCorePoolSize()
                    + ", largest:" + tp.getLargestPoolSize()
                    + ", active:" + tp.getActiveCount()
                    + ", task:" + tp.getTaskCount()
                    + ", service port: " + port);
        }
    }
    return msg.length() == 0 ? new Status(Status.Level.UNKNOWN) : new Status(level, msg.toString());
}

@Override
public void setStartStopThreads(int startStopThreads) {
    this.startStopThreads = startStopThreads;

    // Use local copies to ensure thread safety
    ThreadPoolExecutor executor = startStopExecutor;
    if (executor != null) {
        int newThreads = getStartStopThreadsInternal();
        executor.setMaximumPoolSize(newThreads);
        executor.setCorePoolSize(newThreads);
    }
}

public void start() throws LifecycleException {
    lifecycle.fireLifecycleEvent(BEFORE_START_EVENT, null);
    TaskQueue taskqueue = new TaskQueue();
    TaskThreadFactory tf = new TaskThreadFactory(namePrefix);
    lifecycle.fireLifecycleEvent(START_EVENT, null);
    executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), maxIdleTime, TimeUnit.MILLISECONDS,taskqueue, tf);
    taskqueue.setParent( (ThreadPoolExecutor) executor);
    lifecycle.fireLifecycleEvent(AFTER_START_EVENT, null);
}

public XExecutor getExecutor() {
    if (executor == null) {
        synchronized (DownloadThreadPool.class) {
            if (executor == null) {
                executor = new XExecutor(corePoolSize, MAX_POOL_SIZE, KEEP_ALIVE_TIME, UNIT, //
                                         new PriorityBlockingQueue<Runnable>(),   //无限容量的缓冲队列
                                         Executors.defaultThreadFactory(),        //线程创建工厂
                                         new ThreadPoolExecutor.AbortPolicy());   //继续超出上限的策略，阻止
            }
        }
    }
    return executor;
}

private NetworkTaskScheduler() {
    mExecutor = new ThreadPoolExecutor(
            1,
            1,
            5,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(),
            new NamedThreadFactory(NetworkTaskScheduler.class.getSimpleName())
    );
}

@Override
protected ExecutorService initializeExecutor(
        ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

    BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);
    ThreadPoolExecutor executor  = new ThreadPoolExecutor(
            this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
            queue, threadFactory, rejectedExecutionHandler);
    if (this.allowCoreThreadTimeOut) {
        executor.allowCoreThreadTimeOut(true);
    }

    this.threadPoolExecutor = executor;
    return executor;
}

private ExecutorService createExecutor(Config config) {
    ThreadPoolExecutor service = new ThreadPoolExecutor(config.core, config.max, config.timeout,
            TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(QUEUE_INIT_CAPACITY),
            new TaskThreadFactory(name), new ThreadPoolExecutor.DiscardPolicy());

    allowCoreThreadTimeOut(service, config.allowCoreTimeOut);

    return service;
}