Basic Of Concurrency(十九: 线程池)

线程池是并发编程比较典型的知识点了.当你想在应用中限制同一时间并行执行线程数量的时候可使用线程池.对比经过建立新线程的方式线程池有显著的性能提高且每个线程都能得到必定大小的内存空间用做线程栈存储.

能够将任务经过传入线程池来代替建立新线程的方式让任务并行执行.当线程池有闲置线程时,任务将会被分配给这部分线程执行.在线程池内部任务经过插入阻塞队列的方式来让闲置线程取出执行.当有新任务到达时,将会由一个闲置线程成功取出后执行.其余闲置线程会在线程池中阻塞待命,直到有新的任务到达.

线程池经常使用在多线程服务器上.每个连接经过网络到达服务器时都会被看成一个任务进入到线程池里.线程池中的线程将会并行处理连接中的请求.

Java5的java.util.concurrent包中已经有现成的实现,所以咱们不须要本身手动实现线程池.但咱们仍然颇有必要知道它的底层实现细节.

这里提供了一个简单的线程池实现.你会发现咱们使用了前文阻塞队列中的实现.然而在正式环境中,咱们可使用Java的内建阻塞队列来替代.

public class ThreadPool {
    private BlockingQueue<Runnable> taskQueue;
    private List<PoolThread> threads = new ArrayList<>();
    private boolean isStopped = false;

    public ThreadPool(int initThreads, int maxTasks) {
        taskQueue = new BlockingQueue<>(maxTasks);
        IntStream.range(0, initThreads)
                .mapToObj(i -> new PoolThread(taskQueue))
                .forEach(threads::add);
        threads.forEach(PoolThread::start);
    }

    public synchronized void execute(Runnable task) {
        if (isStopped) throw new IllegalStateException("ThreadPool is stopped");
        taskQueue.enqueue(task);
    }

    public synchronized void stop() {
        isStopped = true;
        threads.forEach(PoolThread::doStop);
    }

    private class PoolThread extends Thread {
        private BlockingQueue<Runnable> taskQueue;
        private boolean isStopped = false;

        public PoolThread(BlockingQueue<Runnable> taskQueue) {
            this.taskQueue = taskQueue;
        }

        @Override
        public void run() {
            while (!isStopped()) {
                try {
                    Runnable runnable = taskQueue.dequeue();
                    runnable.run();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }

        public synchronized void doStop() {
            isStopped = true;
            Thread.currentThread().interrupt(); // break pool thread of dequeue() call.
        }

        public synchronized boolean isStopped() {
            return isStopped;
        }
    }
}
复制代码

一个线程池的实现有两部分.一个是ThreadPool.class来提供对外的入口.而PoolThread则是继承了Thread用于建立线程池中的线程来调度任务.

一个任务能够经过ThreadPool实例的execute(Runnable task)来传入.execute()接受一个Runnable做为参数.而Runnable则是任务的载体,任务的执行代码能够写在Runnable的run方法里.Runnable在线程池内部会被添加到阻塞队列中,等待调度.

阻塞队列中的Runnable会被闲置的PoolThead取出和执行.你能够注意到PoolThread实例中的run()方法.队列为空时线程会阻塞在dequeue()调用上,直到有新的任务进入队列线程才被唤醒,从而取出任务执行,这个过程会持续到线程中止为止.

中止线程池能够经过ThreadPool实例的stop()方法.咱们会注意到stop()方法内部将isStopped变量置换为true且依次调用线程池中的每个线程的doStop()方法.当调用stop()完成后,再次调用execute()方法将会抛出IllegalStateException异常.

线程会在并行执行完任务后中止.咱们注意到PoolThread实例doStop()方法中的Thread.currentThread().interrupt()调用.这可以让阻塞在taskQueue.dequeue()中wait()调用上的线程退出wait()调用,从而退出dequeue()方法抛出一个InterruptedException异常.这个异常会在PoolThread实例的run()方法中捕获,从而从新进入while(!isStopped)检查,从而退出run()方法结束线程.

该系列博文为笔者复习基础所著译文或理解后的产物,复习原文来自Jakob Jenkov所著Java Concurrency and Multithreading Tutorial

上一篇: 阻塞队列
下一篇: Compare And Swap