Question

线程应用实例--简单线程池实现

Answer 1

        对于服务端的程序，经常面对的是客户端传入的短小（执行时间短、工作内容较为单一）任务，需要服务端快速处理并返回结果。如果服务端每次接受到一个任务，创建一个线程，然后进行执行，这在原型阶段是个不错的选择，但是面对成千上万的任务递交进服务器时，如果还是采用一个任务一个线程的方式，那么将会创建数以万记的线程，这不是一个好的选择。因为这会使操作系统频繁的进行线程上下文切换，无故增加系统的负载，而线程的创建和消亡都是需要耗费系统资源的，也无疑浪费了系统资源。

        线程池技术能够很好地解决这个问题，它预先创建了若干数量的线程，并且不能由用户直接对线程的创建进行控制，在这个前提下重复使用固定或较为固定数目的线程来完成任务的执行。这样做的好处是，一方面，消除了频繁创建和消亡线程的系统资源开销，另一方面，面对过量任务的提交能够平缓的劣化。

public interface ThreadPool <Job extends Runnable>{

        //向线程池提交任务

        void execute(Job job);

        //关闭线程池

        void shutdown();

        //增加工作者线程

        void addWorkers(int num);

        //减少工作者线程

        void removeWorker(int num);

        //获取等待执行的任务数

        int getJobSize();

}

public class DefaultThreadPool<Job extends Runnable> implements ThreadPool<Job>{

        //最大工作者线程数量

        private static final int MAX_WORKER_NUMBERS = 10;

        //默认工作者线程数量

        private static final int DEFAULT_WORKER_NUMBERS = 5;

        //最小工作者线程数量

        private static final int MIN_WORKER_NUMBERS = 1;

        //工作任务列表

        private final LinkedList<Job> jobs = new LinkedList<Job>();

        //工作者线程列表

        private final List<Worker> workers = Collections.synchronizedList(new ArrayList<Worker>());

        //工作者线程数量

        private int workerNum = DEFAULT_WORKER_NUMBERS;

        //线程编号生成

        private AtomicLong threadNum = new AtomicLong();

        public DefaultThreadPool() {

                //初始化并启动工作者线程

                initializeWokers(DEFAULT_WORKER_NUMBERS);

        }

        public DefaultThreadPool(int num) {

                workerNum = num > MAX_WORKER_NUMBERS? MAX_WORKER_NUMBERS : num < MIN_WORKER_NUMBERS?                 MIN_WORKER_NUMBERS : num;

                initializeWokers(workerNum);

        }

        //向线程池提交任务֪

        public void execute(Job job) {

                if (job != null) {

                    //提交任务后，向等待的在工作任务列表的线程发送唤醒通知֪

                    synchronized (jobs) {

                            jobs.addLast(job);

                            jobs.notify();

                    }

                }

        }

        public void shutdown() {

                for (Worker worker : workers) {

                    worker.shutdown();

                }

        }

        //增加工作者线程

        public void addWorkers(int num) {

                synchronized (jobs) {

                        if (num + this.workerNum > MAX_WORKER_NUMBERS) {

                                num = MAX_WORKER_NUMBERS - this.workerNum;

                        }

                        initializeWokers(num);

                        this.workerNum += num;

                }

        }

        //减少工作者线程

        public void removeWorker(int num) {

                synchronized (jobs) {

                        if (num >= this.workerNum) {

                                throw new IllegalArgumentException("beyond workNum");

                        }

                        int count = 0;

                        while (count < num) {

                                Worker worker = workers.get(count);

                                if (workers.remove(worker)) {

                                        worker.shutdown();

                                        count++;

                                }

                        }

                        this.workerNum -= count;

                }

        }

        //获取等待执行的任务数

        public int getJobSize() {

                    return jobs.size();

        }

        //初始化工作者线程列表

        private void initializeWokers(int num) {

                for (int i = 0; i < num; i++) {

                    Worker worker = new Worker();

                    workers.add(worker);

                    Thread thread = new Thread(worker, "ThreadPool-Worker-" + threadNum.

                    incrementAndGet());

                    thread.start();

            }

        }

        //工作者线程定义

        class Worker implements Runnable {

               //是否工作

                private volatile boolean running = true;

                public void run() {

                        while (running) {

                            Job job = null;

                            synchronized (jobs) {

                                    //如果任务列表中没有任务则等待

                                    while (jobs.isEmpty()) {

                                            try {

                                                    jobs.wait();

                                             } catch (InterruptedException ex) {

                                                    Thread.currentThread().interrupt();

                                                    return;

                                             }

                                    }

                                    //获取任务执行

                                    job = jobs.removeFirst();

                                    if(job != null) {

                                        try {

                                            job.run();

                                        } catch (Exception ex) {

                                        }

                                    }

                        }

                }

            }

            public void shutdown() {

                    running = false;

                }

        }

}

        当客户端调用execute(Job)方法时，会不断地向任务列表jobs中添加Job，而每个工作者线程当客户端调用execute(Job)方法时，会不断地向任务列表jobs中添加Job，而每个工作者线程会不断地从jobs上取出一个job进行执行，当jobs为空时，工作者线程进入等到状态。

        添加一个job后，对工作队列jobs调用了notify()方法，而不是notifyAll()方法，因为能够确定有工作者线程被唤醒，这时使用notify()方法将会比notifyAll()方法获得更小的开销（避免将等待队列中的线程全部移到阻塞队列中）。

        可以看到线程池的本质是使用了一个线程安全的工作队列连接工作者线程和客户端线程，客户端线程将任务放入工作队列后边返回，而工作者线程则不断地从工作任务队列取出工作并执行。当工作队列为空时，所有的工作者线程均等待在工作队列上，当有客户端提交了一个任务之后会通知任意一个工作者线程，随着大量任务的提交，更多的工作者线程会被唤醒。