Java多线程高并发高级篇(四)--线程池工作流程篇

zhaodengfeng1989

浏览: 69774 次
性别:
来自: 北京

最近访客更多访客>>

kongqingshuai

陈睿mikechen

tom2139779

石晓楠

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

线程池

前三篇说了线程池一些核心东西，这一篇，我想就线程池的使用从全流程的角度来说下，线程池是如何工作的。

我们以一个调用示例，来逐步深入。

一、固定大小线程池调用示例

1、我们先创建一个corePoolSize大小为5的线程池，然后提交10个任务。

public class ThreadPoolDemo {
	
	public static class MyTask implements Runnable{

		@Override
		public void run() {
			// TODO Auto-generated method stub
			System.out.println(System.currentTimeMillis()+"--Thread Id:"+Thread.currentThread().getId());
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		
	}
	
	public static void main(String[] args) {
		MyTask task = new MyTask();
		//定义corePoolSize和maxiumPoolSize大小为5的线程池
		ExecutorService es = Executors.newFixedThreadPool(5);
		for (int i = 0; i < 10; i++) {
			es.execute(task);
		}
	}
}

首先，我们要知道的是，在示例中，我们初始化的线程池内容是什么样子的。我们使用的是大小固定的线程池Executors.newFixedThreadPool(5)。

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

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

初始化完成，各属性值见下图，清晰明白，其他不多说。

注意两个属性（要不然看源码都不知道找谁）：threadFactory为Executors中的静态内部类DefaultThreadFactory；handler（拒绝策略处理器）使用的是默认的ThreadPoolExecutor中的静态内部类AbortPolicy。

2、我们从es.execute(task)开始分析整个的执行流程。execute方法是ThreadPoolExecutor中的方法，源码内容如下：

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
            if (runState == RUNNING && workQueue.offer(command)) {
                if (runState != RUNNING || poolSize == 0)
                    ensureQueuedTaskHandled(command);
            }
            else if (!addIfUnderMaximumPoolSize(command))
                reject(command); // is shutdown or saturated
        }
    }

①初始时，poolSize值为0，条件poolSize >= corePoolSize不成立，走的都是方法addIfUnderCorePoolSize(command)，其中command是提交的需要执行的任务。

该方法源码如下：

/**
     * Creates and starts a new thread running firstTask as its first
     * task, only if fewer than corePoolSize threads are running
     * and the pool is not shut down.
     * @param firstTask the task the new thread should run first (or
     * null if none)
     * @return true if successful
     */
    private boolean addIfUnderCorePoolSize(Runnable firstTask) {
        Thread t = null;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (poolSize < corePoolSize && runState == RUNNING)
                t = addThread(firstTask);
        } finally {
            mainLock.unlock();
        }
        if (t == null)
            return false;
        t.start();
        return true;
    }

核心代码中关键方法为：addThread(firstTask)，源码如下：

/**
     * Creates and returns a new thread running firstTask as its first
     * task. Call only while holding mainLock.
     *
     * @param firstTask the task the new thread should run first (or
     * null if none)
     * @return the new thread, or null if threadFactory fails to create thread
     */
    private Thread addThread(Runnable firstTask) {
        Worker w = new Worker(firstTask);
        Thread t = threadFactory.newThread(w);
        if (t != null) {
            w.thread = t;
            workers.add(w);
            int nt = ++poolSize;
            if (nt > largestPoolSize)
                largestPoolSize = nt;
        }
        return t;
    }

它的作用是在线程池中的线程数量（poolsize）还没有达到指定的corePoolSize之前，将任务提交给线程池中新创建的线程处理（Worker w = new Worker(firstTask);Thread t = threadFactory.newThread(w);），这里做了一层封装，把任务包装成Worker，然后使用Work对象在threadFactory创建新的线程，也就是使用这个线程去处理work。由于这个过程都能创建新线程处理任务，所以if (t == null)不成立,返回的是true,所以条件 if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command))不成立,下面的逻辑不会执行（这也是设计巧妙的地方）。其他逻辑很简单（每创建一个新线程，poolsize都会自增1，统计目前线程池的大小）。

②当线程池大小达到corePoolSize时，也就是条件poolSize >= corePoolSize成立，不会再执行addIfUnderCorePoolSize方法，因为只要线程池不关闭，接收新任务和处理等待队列的任务过程就不会停，所以runState在这个过程始终是0，也就是RUNNING状态（RUNNING-0: Accept new tasks and process queued tasks）。

if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
            if (runState == RUNNING && workQueue.offer(command)) {
                if (runState != RUNNING || poolSize == 0)
                    ensureQueuedTaskHandled(command);
            }
            else if (!addIfUnderMaximumPoolSize(command))
                reject(command); // is shutdown or saturated
        }

左半条件成立后，那么接着就要进行 workQueue.offer(command)了。等待任务队列（workQueue）这回就要派上用场了。因为固定大小的线程池使用的等待任务队列是LinkedBlockingQueue，所以我们看下它的offer方法，核心代码就一行-----insert（e），只要插入成功，就返回true（c变为0，c>=0肯定是成立的）。其实跟我们前面介绍队列术时候的put方法一样，就是放任务，前面帖子已经说过，这里不再赘述，不清楚的可以回过去看。

public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        final AtomicInteger count = this.count;
        if (count.get() == capacity)
            return false;
        int c = -1;
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            if (count.get() < capacity) {
                insert(e);
                c = count.getAndIncrement();
                if (c + 1 < capacity)
                    notFull.signal();
            }
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
        return c >= 0;
    }

我们使用固定大小线程池，涉及到核心方法只有addIfUnderCorePoolSize，因为固定线程池使用的等待任务队列是LinkedBlockingQueue，接近于无限队列，因此在没有达到Integer.MAX_VALUE个任务时，是不会执行到addIfUnderMaximumPoolSize方法的。为了看下addIfUnderMaximumPoolSize做了什么，我们还需要举一例，使用newCachedThreadPool。

二、newCachedThreadPool使用举例

为什么要用它作为举例，因为它有个特殊之处，workQueue是使用SynchronousQueue，无大小，也就是人们说的直接提交队列（不明白的回头看我写的队列术解密），因此很容易走到方法addIfUnderMaximumPoolSize（if under maximun pool size，add command）。

我们看下它的源码：

private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {
        Thread t = null;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (poolSize < maximumPoolSize && runState == RUNNING)
                t = addThread(firstTask);
        } finally {
            mainLock.unlock();
        }
        if (t == null)
            return false;
        t.start();
        return true;
    }

这段源码，核心代码也就一段：

 if (poolSize < maximumPoolSize && runState == RUNNING)
                t = addThread(firstTask);

当等待任务队列已满时，那么就需要判断线程池中处理任务的线程个数是否达到了规定的最大线程池大小maxinumPoolSize，如果没有达到，则将任务使用work对象包裹任务，然后利用work对象创建新线程至线程池中，处理该任务；如果线程池大小达到了maxinumPoolSize，那么就需要执行reject（command）--执行拒绝策略了。

附示例及示例程序执行过程图：

public static void main(String[] args) {
		MyTask task = new MyTask();
		//定义corePoolSize和maxiumPoolSize大小为5的线程池
		ExecutorService es = Executors.newCachedThreadPool();
		for (int i = 0; i < 10; i++) {
			es.execute(task);
		}
	}

执行过程程序图：