初窥Java线程池

如何创建一个线程（两种方式）

方式一：继承Thread类，覆写run方法

public class Test {

    public static void main(String[] args) {

        Thread thread = new ThreadDemo();
        thread.start();
    }

    private static class ThreadDemo extends Thread {

        @Override
        public void run() {
            System.out.println("This is a thread :" + Thread.currentThread().getId());
        }
    }
}

方式二：实现Runnable接口（推荐使用这种方式，扩展性更强）

public class Test {

    public static void main(String[] args) {

        Thread thread = new Thread(new Task());
        thread.start();
    }

    private static class Task implements Runnable {

        @Override
        public void run() {
            System.out.println("This is a thread :" + Thread.currentThread().getId());
        }
    }
}

为什么要使用线程池？

使用线程池，可以大大提高任务处理效率，因为线程的创建与销毁都需要消耗一定的资源，如果频繁的创建与销毁线程，不仅损耗可观的系统资源，而且响应慢；但是如果我们预先初始化一定数量线程，并且在使用完的时候不立即销毁（保存一定时间），这样就可以高响应外部任务（不需要等待），并且大大提高线程利用率（同数据库连接池一个道理）;

说了线程池的好处，我们如何使用线程池呢？(需要我们自己写一个?)

嘿嘿，真不用，java已经给我们提供了一个非常棒的线程池工具(Executors)，使用它你可以非常简单的控制，管理线程；

下面我们看下Excutors中的几个常用方法：

// 创建具有固定数量的线程池(线程池中线程数量始终固定为nThreads)
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}


// 创建只有一个线程的线程池（其实就是Threads退化为1）
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}


// 对于每一个任务总是用一个线程去执行(不缓冲)，并且空闲线程自动保留60秒
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

发现没？其实这些方法都是封装ThreadPoolExecutor类而已,那我们现在来看看ThreadPoolExecutor类的构造方法：

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

重点关注这几个参数：

corePoolSize：	核心线程数(当线程池中的线程数目小于corePoolSize时，对于提交过来的任务总是起一个新的线程去运行);
maximumPoolSize：	线程池允许的最大线程数;
keepAliveTime：	当线程数大于corePoolSize时，那些空闲线程保留时间;
workQueue：	工作队列（当线程池中的线程数目大于等于corePoolSize时，对于提交的任务总是放在工作对列，只有当工作队列满的时候，才会新起一个线程去执行)
handler：	异常处理策略(当工作队列已满，线程池数目也已达到maximumPoolSize，这时候对于提交过来任务的处理策略(有4种，稍后我会代码详述));

总结一下ThreadPoolExecutor处理任务的过程就是【important】：

若线程池中的线程数目小于corePoolSize,这时候总是用一个新的线程去执行提交的任务，而不缓存；

若线程池中的数目大于等于corePoolSize,这时候任务总是先被放入工作队列，只有当工作队列满的时候，才会新起一个线程去执行；

若工作队列已经满了，并且线程池中的数目也达到了maximumPoolSize,这时候对于提交过来的任务，只能执行异常策略了（可能被丢弃，也可能由提交任务的线程来执行（这种策略可以减缓任务提交速度））；

// 上述逻辑相关代码：
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
    }
}

再补充几句：

如果使用了LinkedBlockingQueue(无界队列)，由于队列始终不会满（除非内存资源耗尽），线程池中的线程数目会始终为corePoolSize（多出的任务会放在工作队列里）,那么所以maximumPoolSize和keepAlive这两个参数实际上是没有意义的；

若使用ArrayBlockingQueue(有界队列，它最能体现线程池的工作流程，其余都是临界情况)，你需要权衡线程数和工作队列大小；

至于SynchronousQueue，自己不是很清楚，T T：


最后我们详细说一下线程池的异常处理策略：

测试代码：

public class Test {

    public static void main(String[] args) {

    	 /**
         * corePoolSize:2
         * maximumPoolSize:3
         * queueSize:1
         * keepAlive:5s
         */
        ExecutorService executors = new ThreadPoolExecutor(2, 3, 5, TimeUnit.SECONDS,
                                                           new ArrayBlockingQueue<Runnable>(1),
                                                           new ThreadPoolExecutor.AbortPolicy());

        System.out.println("main thread: " + Thread.currentThread().getId());

        for (int i = 1; i <= 10; i++) {
            final int taskId = i;

            executors.execute(new Runnable() {

                // 每一个task需要执行1min
                public void run() {
                    System.out.println("start thread: " + Thread.currentThread().getId() + "--task id: " + taskId);
                    try {
                        TimeUnit.SECONDS.sleep(60);
                    } catch (InterruptedException e) {
                    }
                    System.out.println("end thread: " + Thread.currentThread().getId() + "--task id: " + taskId);
                }
            });
        }
        executors.shutdown();
    }
}

1.AbortPolicy策略(这是默认策略，该策略会丢弃任务，并抛出异常)

我们看下执行结果

main thread: 1
start thread: 11--task id: 1
start thread: 12--task id: 2
start thread: 13--task id: 4
Exception in thread "main" java.util.concurrent.RejectedExecutionException

对于task1,task2 总是用一个新的线程去执行(此时线程数目已经达到corePoolSize),这时候又来了一个task3, 于是放入了工作队列（此时工作队列也满了），接着又来了一个task4(好吧，由于队列满了，只能再起一个线程去执行了，此时线程数已经达到maximumPoolSize),可谁知又来了一个task5（工作队列满了，线程数也已经达到maximumPoolSize），这时候只能丢弃任务并抛异常(RejectedExecutionException);

2.DiscardPolicy(丢弃任务但不抛异常)

我们看下执行结果：

main thread: 1
start thread: 11--task id: 1
start thread: 12--task id: 2
start thread: 13--task id: 4
end thread: 12--task id: 2
start thread: 12--task id: 3
end thread: 11--task id: 1
end thread: 13--task id: 4
end thread: 12--task id: 3

对于task1,task2用一个新的线程去执行，task3被放入队列，task4又新起一个线程去执行（因为工作队列已经满了），其余task都被丢弃(不抛异常)，当task2执行完毕（有空闲线程了),于是从工作队列中取出task3去执行；

3.CallerRunsPolicy(由提交任务的线程去执行，该策略可以保证任务不被丢弃)

我们看下执行结果：

main thread: 1
start thread: 11--task id: 1
start thread: 12--task id: 2
start thread: 13--task id: 4
start thread: 1--task id: 5    ---> important
...... 省略

之前策略对于task5总是丢弃的，但是对于CallerRunsPolicy策略，它会让提交任务的线程去执行task(即这里的main thread)


4.DiscardOldestPolicy(删除队列中老的任务(即头元素))

我们看下执行结果：

main thread: 1
start thread: 11--task id: 1
start thread: 12--task id: 2
start thread: 13--task id: 4
end thread: 11--task id: 1
start thread: 11--task id: 10  ---> important
end thread: 12--task id: 2
end thread: 13--task id: 4
...... 省略

工作队列中任务总是被后来的任务所替代(队列大小为1)，所以最后工作队列中只有最后一个任务task10


写到这里，我想大家对异常策略应该比较了解了！


参考文档：
http://dongxuan.iteye.com/blog/901689
作者写的很棒，看了获益匪浅，但是关于SynchronousQueue与作者理解的有所同...