【Java线程】线程池的原理和实现

柔光的暖阳◎ 2022-08-21 08:56 210阅读 0赞

### **1、为什么要使用线程池？** ###

线程池是Java5提供的一个新技术，方便我们快速简洁的定义线程池。

诸如 Web 服务器、数据库服务器、文件服务器或邮件服务器之类的许多服务器应用程序都面向处理来自某些远程来源的大量短小的任务。请求以某种方式到达服务器，这种方式可能是通过网络协议（例如 HTTP、FTP 或 POP）、通过 JMS 队列或者可能通过轮询数据库。不管请求如何到达，服务器应用程序中经常出现的情况是：单个任务处理的时间很短而请求的数目却是巨大的。

构建服务器应用程序的一个过于简单的模型应该是：每当一个请求到达就创建一个新线程，然后在新线程中为请求服务。实际上，对于原型开发这种方法工作得很好，但如果试图部署以这种方式运行的服务器应用程序，那么这种方法的严重不足就很明显。每个请求对应一个线程（thread-per-request）方法的不足之一是：为每个请求创建一个新线程的开销很大；为每个请求创建新线程的服务器在创建和销毁线程上花费的时间和消耗的系统资源要比花在处理实际的用户请求的时间和资源更多。

除了创建和销毁线程的开销之外，活动的线程也消耗系统资源。在一个 JVM 里创建太多的线程可能会导致系统由于过度消耗内存而用完内存或“切换过度”。为了防止资源不足，服务器应用程序需要一些办法来限制任何给定时刻处理的请求数目。

**线程池为线程生命周期开销问题和资源不足问题提供了解决方案。通过对多个任务重用线程，线程创建的开销被分摊到了多个任务上。其好处是，因为在请求到达时线程已经存在，所以无意中也消除了线程创建所带来的延迟。这样，就可以立即为请求服务，使应用程序响应更快。而且，通过适当地调整线程池中的线程数目，也就是当请求的数目超过某个阈值时，就强制其它任何新到的请求一直等待，直到获得一个线程来处理为止，从而可以防止资源不足**。

### **2、线程池的实现原理** ###

**多线程技术主要解决处理器单元内多个线程执行的问题，它可以显著减少处理器单元的闲置时间，增加处理器单元的吞吐能力**。

假设一个服务器完成一项任务所需时间为：T1 创建线程时间，T2 在线程中执行任务的时间，T3 销毁线程时间。

如果：T1 + T3 远大于 T2，则可以采用线程池，以提高服务器性能。

一个线程池包括以下四个基本组成部分：

> 1、线程池管理器（ThreadPool）：用于创建并管理线程池，包括 创建线程池，销毁线程池，添加新任务；
> 
> 2、工作线程（PoolWorker）：线程池中线程，在没有任务时处于等待状态，可以循环的执行任务；
> 
> 3、任务接口（Task）：每个任务必须实现的接口，以供工作线程调度任务的执行，它主要规定了任务的入口，任务执行完后的收尾工作，任务的执行状态等；
> 
> 4、任务队列（taskQueue）：用于存放没有处理的任务。提供一种缓冲机制。

线程池技术正是关注如何**缩短或调整T1,T3时间**的技术，从而提高服务器程序性能的。它把T1，T3分别安排在服务器程序的启动和结束的时间段或者一些空闲的时间段，这样在服务器程序处理客户请求时，不会有T1，T3的开销了。

线程池不仅调整T1,T3产生的时间段，而且它还显著减少了创建线程的数目，看一个例子：

> 假设一个服务器一天要处理50000个请求，并且每个请求需要一个单独的线程完成。在线程池中，线程数一般是固定的，所以产生线程总数不会超过线程池中线程的数目，而如果服务器不利用线程池来处理这些请求则线程总数为50000。一般线程池大小是远小于50000。所以利用线程池的服务器程序不会为了创建50000而在处理请求时浪费时间，从而提高效率。
> 
> 代码实现中并没有实现任务接口，而是把Runnable对象加入到线程池管理器（ThreadPool），然后剩下的事情就由线程池管理器（ThreadPool）来完成了

/** 
     * 线程管理器：创建线程，执行任务，销毁线程，获取线程基本信息 
     */  
    public final class ThreadPool {
          
        // 线程池中默认线程的个数为5  
        private static int worker_num = 5;  
        // 工作线程  
        private WorkThread[] workThrads;  
        // 已完成的任务个数  
        private static volatile int finished_task = 0;  
        // 任务队列，作为一个缓冲,List线程不安全  
        private List<Runnable> taskQueue = new LinkedList<Runnable>();  
        private static ThreadPool threadPool;  
    
        // 创建具有默认线程个数的线程池  
        private ThreadPool() {  
            this(5);  
        }  
    
        // 创建线程池,worker_num为线程池中工作线程的个数  
        private ThreadPool(int worker_num) {  
            ThreadPool.worker_num = worker_num;  
            workThrads = new WorkThread[worker_num];  
            for (int i = 0; i < worker_num; i++) {  
                workThrads[i] = new WorkThread();  
                workThrads[i].start();// 开启线程池中的线程  
            }  
        }  
    
        // 单态模式，获得一个默认线程个数的线程池  
        public static ThreadPool getThreadPool() {  
            return getThreadPool(ThreadPool.worker_num);  
        }  
    
        // 单态模式，获得一个指定线程个数的线程池,worker_num(>0)为线程池中工作线程的个数  
        // worker_num<=0创建默认的工作线程个数  
        public static ThreadPool getThreadPool(int worker_num1) {  
            if (worker_num1 <= 0)  
                worker_num1 = ThreadPool.worker_num;  
            if (threadPool == null)  
                threadPool = new ThreadPool(worker_num1);  
            return threadPool;  
        }  
    
        // 执行任务,其实只是把任务加入任务队列，什么时候执行有线程池管理器觉定  
        public void execute(Runnable task) {  
            synchronized (taskQueue) {  
                taskQueue.add(task);  
                taskQueue.notify();  
            }  
        }  
    
        // 批量执行任务,其实只是把任务加入任务队列，什么时候执行有线程池管理器觉定  
        public void execute(Runnable[] task) {  
            synchronized (taskQueue) {  
                for (Runnable t : task)  
                    taskQueue.add(t);  
                taskQueue.notify();  
            }  
        }  
    
        // 批量执行任务,其实只是把任务加入任务队列，什么时候执行有线程池管理器觉决定 
        public void execute(List<Runnable> task) {  
            synchronized (taskQueue) {  
                for (Runnable t : task)  
                    taskQueue.add(t);  
                taskQueue.notify();  
            }  
        }  
    
        // 销毁线程池,该方法保证在所有任务都完成的情况下才销毁所有线程，否则等待任务完成才销毁  
        public void destroy() {  
            while (!taskQueue.isEmpty()) {
       // 如果还有任务没执行完成，就先睡会吧  
                try {  
                    Thread.sleep(10);  
                } catch (InterruptedException e) {  
                    e.printStackTrace();  
                }  
            }  
            // 工作线程停止工作，且置为null  
            for (int i = 0; i < worker_num; i++) {  
                workThrads[i].stopWorker();  
                workThrads[i] = null;  
            }  
            threadPool=null;  
            taskQueue.clear();// 清空任务队列  
        }  
    
        // 返回工作线程的个数  
        public int getWorkThreadNumber() {  
            return worker_num;  
        }  
    
        // 返回已完成任务的个数,这里的已完成是只出了任务队列的任务个数，可能该任务并没有实际执行完成  
        public int getFinishedTasknumber() {  
            return finished_task;  
        }  
    
        // 返回任务队列的长度，即还没处理的任务个数  
        public int getWaitTasknumber() {  
            return taskQueue.size();  
        }  
    
        // 覆盖toString方法，返回线程池信息：工作线程个数和已完成任务个数  
        @Override  
        public String toString() {  
            return "WorkThread number:" + worker_num + "  finished task number:"  
                    + finished_task + "  wait task number:" + getWaitTasknumber();  
        }  
    
        /** 
         * 内部类，工作线程 
         */  
        private class WorkThread extends Thread {
          
            // 该工作线程是否有效，用于结束该工作线程  
            private boolean isRunning = true;  
    
            /* 
             * 关键所在啊，如果任务队列不空，则取出任务执行，若任务队列空，则等待 
             */  
            @Override  
            public void run() {  
                Runnable r = null;  
                while (isRunning) {
       // 注意，若线程无效则自然结束run方法，该线程就没用了  
                    synchronized (taskQueue) {  
                        while (isRunning && taskQueue.isEmpty()) {
       // 队列为空  
                            try {  
                                taskQueue.wait(20);  
                            } catch (InterruptedException e) {  
                                e.printStackTrace();  
                            }  
                        }  
                        if (!taskQueue.isEmpty())  
                            r = taskQueue.remove(0);// 取出任务  
                    }  
                    if (r != null) {  
                        r.run();// 执行任务  
                    }  
                    finished_task++;  
                    r = null;  
                }  
            }  
    
            // 停止工作，让该线程自然执行完run方法，自然结束  
            public void stopWorker() {  
                isRunning = false;  
            }  
        }  
    }

测试一下：

public class TestThreadPool {
          
        public static void main(String[] args) {  
            // 创建3个线程的线程池  
            ThreadPool t = ThreadPool.getThreadPool(3);  
            t.execute(new Runnable[] { new Task(), new Task(), new Task() });  
            t.execute(new Runnable[] { new Task(), new Task(), new Task() });  
            System.out.println(t);  
            t.destroy();// 所有线程都执行完成才destory  
            System.out.println(t);  
        }  
    
        // 任务类  
        static class Task implements Runnable {
          
            private static volatile int i = 1;  
    
            @Override  
            public void run() {
       // 执行任务  
                System.out.println("任务 " + (i++) + " 完成");  
            }  
        }  
    }

运行结果：

WorkThread number:3  finished task number:0  wait task number:6
    任务 1 完成
    任务 2 完成
    任务 3 完成
    任务 4 完成
    任务 5 完成
    任务 6 完成
    WorkThread number:3  finished task number:6  wait task number:0

### **3、Java5提供的线程池** ###

![这里写图片描述][SouthEast]

![这里写图片描述][SouthEast 1]

#### **3.1 缓存线程池(newCachedThreadPool)** ####

缓存线程池(newCachedThreadPool)可以创建任意个线程，每个任务过来后都会创建一个线程，用于**任务少，或执行时间短的任务**，例如我们创建十个任务，那么缓冲线程池将会创建十个线程来执行。如下代码：

ExecutorService threadPool = Executors.newCachedThreadPool();  
    
    for(int i=1; i<=10; i++){
        final int taskId = i;  
            threadPool.execute(new Runnable(){  
                 public void run() {  
                      for(int i=1; i<=10; i++){  
                          System.out.println(Thread.currentThread().getName() + " is looping of " + i + " the task is " + taskId);  
                          try {  
                              Thread.sleep(20);  
                          } catch (InterruptedException e) {  
                              // TODO Auto-generated catch block  
                              e.printStackTrace();  
                          }  
                      }  
                 }  
           });  
    }  
    System.out.println("add  all of 10 task");  
    threadPool.shutdown();

#### **3.2 固定数量线程池（newFixedThreadPool）** ####

固定数量线程池（newFixedThreadPool）允许我们创建固定线程数量的线程池，如果任务数大于线程池中线程的数量，那么任务将等待，如下代码：

ExecutorService threadPool = Executors.newFixedThreadPool(3);         
    for(int i=1; i<=10; i++){  
        final int taskId = i;  
        threadPool.execute(new Runnable(){  
    
            public void run() {  
                for(int i=1; i<=10; i++){  
                    System.out.println(Thread.currentThread().getName() + " is looping of " + i + " the task is " + taskId);  
                    try {  
                        Thread.sleep(20);  
                    } catch (InterruptedException e) {  
                        // TODO Auto-generated catch block  
                        e.printStackTrace();  
                    }  
    
                }  
            }  
    
        });  
    }  
    System.out.println("add  all of 10 task");  
    threadPool.shutdown();

#### **3.4 单一线程池(newSingleThreadExecutor)** ####

如何实现线程挂掉后重新启动？创建单一的线程池。

newSingleThreadExecutor()，这样线程池中只会有一个线程工作，当线程失败后会重新创建一个线程将失败的线程替换掉。

3.5 定时器线程池（scheduleAtFixedRate）

定时器线程池（scheduleAtFixedRate）与定时器很类似，可以指定线程池中线程在多长时间后执行，以及每个多长时间执行一次，代码如下，可以模拟让炸弹在6s后爆炸，并且每个2s炸一次：

Executors.newScheduledThreadPool(3).scheduleAtFixedRate(  
    //      .schedule(  
                    new Runnable(){  
    
                public void run() {  
                    System.out.println("boming");  
                }  
    
            }, 6, 2, TimeUnit.SECONDS);  
        }

[SouthEast]: /images/20220724/30469bfd38364d17ba48cae4b0e9d8e4.png
[SouthEast 1]: /images/20220724/5bf7e1dd96b3482d8d29a38da9036a7c.png