【Java并发】线程池的创建

注：《重构与模式》将构造重构成created method，在线程池的创建中隐约看到些影子，故希望能比较下

获得线程池实例

 

1.ThreadPoolExecutor构造

提供了四个构造如下：

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue)

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, 　　　　                    RejectedExecutionHandler handler)

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue,

　　　　　　 ThreadFactory threadFactory)

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue,

　　　　       ThreadFactory threadFactory, RejectedExecutionHandler handler)

 

其中

int corePoolSize：保留在pool中的线程的数目，即使这些线程是idle（空闲）的，也不会terminate。换句话说就是pool会维持corePoolSize数目的线程。

Q:如果有idle thread，那么这些idle thread占用资源情况怎么样？

A:

int maximumPoolSize：pool所容纳线程的最大数目。

long keepAliveTime：当 number of thread > core thread时，this is the maximum time that excess idle threads will wait for new tasks before terminating.对于那些idle thread，会最长等待keepAliveTime时间来获取新任务，如果仍没有新任务的话则terminate。

BlockingQueue<Runnable> workQueue：任务队列，在任务被执行前，workqueue来hold这些任务。该queue只能用来hold那些由execute方法提交的Runnable型的任务。

Q:通过execute提交的任务都会放到任务队列中么？还是当线程池来不及处理execute提交的任务时放到队列里去？

A:

ThreadFactory threadFactory：线程池利用thread factory来创建线程。

RejectedExecutionHandler handler：当任务的执行被blocked（阻塞），会用到RejectedExecutionHandler

Q:何时任务的执行会blocked？

A:the thread bounds and queue capacities are reached。即当线程的数目已经达到maximumPoolSize，并且任务队列也满了。

源码如下：

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
}

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
}

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), handler);
}

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
}

 

 2. Executors静态方法

 

static ExecutorService newFixedThreadPool(int nThreads)

static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory)

static ExecutorService newSingleThreadExecutor()

static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory)

static ExecutorService newCachedThreadPool()

static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) 

【示例代码】

代码1：

 

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

ExecutorService executor = Executors.newFixedThreadPool(maxThread, new ThreadFactory() {
         private AtomicInteger seqNo = new AtomicInteger(1);
         
         public Thread newThread(Runnable r) {
            return new Thread(r,"SMS Log Search Engine Thread -- "+seqNo.getAndIncrement());
         }
      });

 

源码如下

//nThreads ：fixed number of threads，在任何一个时刻，at most只有nThreads个活跃线程在执行任务
public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
 }

public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
 }

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory));
    }
//当执行许多short-lived asynchronous tasks时，能显著提高性能，一个线程如果超过imin没有任务执行，就会terminated并从cache中移除
//so 如果没有任务执行的话，线程池不会占用任何资源
public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

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