线程池系列三：ThreadPoolExecutor讲解

三、一个用队列处理线程池例子

package demo;

import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolExecutorTest
{

    private static int queueDeep = 4;

    public void createThreadPool()
    {
        /*
         * 创建线程池，最小线程数为2，最大线程数为4，线程池维护线程的空闲时间为3秒，
         缓冲队列为4，线程执行时间是3秒。
         */
        ThreadPoolExecutor tpe = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueDeep),
                new ThreadPoolExecutor.DiscardOldestPolicy()); //这里采取的是抛弃旧的任务


        // 向线程池中添加 10 个任务
        for (int i = 0; i < 10; i++)
        {
            try
            {
                Thread.sleep(1);
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
            while (getQueueSize(tpe.getQueue()) >= queueDeep)
            {
                System.out.println("队列已满，等3秒再添加任务");
                try
                {
                    Thread.sleep(3000);
                }
                catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }
            TaskThreadPool ttp = new TaskThreadPool(i);
            System.out.println("put i:" + i);
            tpe.execute(ttp);
        }

        tpe.shutdown();
    }

    private synchronized int getQueueSize(Queue queue)
    {
        return queue.size();
    }

    public static void main(String[] args)
    {
        ThreadPoolExecutorTest test = new ThreadPoolExecutorTest();
        test.createThreadPool();
    }

    class TaskThreadPool implements Runnable
    {
        private int index;

        public TaskThreadPool(int index)
        {
            this.index = index;
        }

        public void run()
        {
            System.out.println(Thread.currentThread() + " index:" + index);
            try
            {
                Thread.sleep(3000);
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }
    }
}

说明：

这里执行的结果为：

put i:0　　　　　　　　　　　　　　　　　　　　　　0加入进去池子
Thread[pool-1-thread-1,5,main] index:0　　　　　0开始执行
put i:1　　　　　　　　　　　　　　　　　　　　　　1加入进去池子
Thread[pool-1-thread-2,5,main] index:1    　　　  1开始执行
put i:2　　　　　　　　　　　　　　　　　　　　　　2加入缓冲
put i:3　　　　　　　　　　　　　　　　　　　　　　3加入缓冲
put i:4　　　　　　　　　　　　　　　　　　　　　　4加入缓冲
put i:5　　　　　　　　　　　　　　　　　　　　　　5加入缓冲
队列已满，等3秒再添加任务

（上述代码首先一次性执行，会在这里停一会，因为添加线程不需要时间）
Thread[pool-1-thread-1,5,main] index:2　　　　2开始执行
Thread[pool-1-thread-2,5,main] index:3　　　　3开始执行
put i:6　　　　　　　　　　　　　　　　　　　　　6加入缓冲
put i:7　　　　　　　　　　　　　　　　　　　　　7加入缓冲
队列已满，等3秒再添加任务

（会停一会，因为任务执行要3秒时间，2,3,4,5会先执行2,3 ，因为2,3先进入缓冲队列）
Thread[pool-1-thread-1,5,main] index:4
Thread[pool-1-thread-2,5,main] index:5
put i:8
put i:9

（停一会，任务执行要3秒，）
Thread[pool-1-thread-1,5,main] index:6
Thread[pool-1-thread-2,5,main] index:7

（要停一会，任务执行要3秒）
Thread[pool-1-thread-1,5,main] index:8
Thread[pool-1-thread-2,5,main] index:9

 （执行完毕）

ps:这里是当队列已满时线程就一直等待了，不会再新创建线程，所以一直就只有1和2两个线程来执行。

如果把

while (getQueueSize(tpe.getQueue()) >= queueDeep){}

 这一段去掉，那么执行结果为：

put i:0
Thread[pool-1-thread-1,5,main] index:0
put i:1
Thread[pool-1-thread-2,5,main] index:1
put i:2
put i:3
put i:4
put i:5
put i:6
Thread[pool-1-thread-3,5,main] index:6
put i:7
Thread[pool-1-thread-4,5,main] index:7
put i:8
put i:9
Thread[pool-1-thread-1,5,main] index:4
Thread[pool-1-thread-2,5,main] index:5
Thread[pool-1-thread-3,5,main] index:8
Thread[pool-1-thread-4,5,main] index:9

ps:这个执行顺序是0，1两个任务先进来，分别由线程1,2来执行，然后2,-5进来，队列满，6任务进来，因为队列已满，且1,2线程还未执行完，没有可用的线程，所以创建新的线程来运行6。7任务同理。然后8任务进来，队列已满，且1,2,3,4线程未执行完，线程数又等于了最多4个线程的限制，这时看线程池的执行策略为DiscardOldestPolicy,就是抛弃旧的任务，故开始进队列的2任务被抛弃，3任务同理，8,9任务进入队列，然后这时1-4线程已经执行完自己的任务，开始执行队列中的4,5,8,9

如果更改执行策略，那么相应的结果也会不一样，如果不希望有任务被抛弃，那么可以采用CallerRunsPolicy()策略。