线程池的创建以及使用

女爷i 2022-02-19 16:14 268阅读 0赞

> **线程池中每个类的具体关系 **

/**
         *
         * 	        线程池各类的关系：
         * 			Executors -- new ThreadPoolExecutor
         *
         * 			ThreadPoolExecutor extends AbstractExecutorService
         *
         * 			AbstractExecutorService(抽象类) implements ExecutorService
         *
         * 			interface ExecutorService extends Executor
         */

> 产生死锁的四种条件以及解决方案

1、互斥：某种资源一次只允许一个进程访问，即该资源一旦分配给某个进程，其他进程就不能再访问，直到该进程访问结束。  
    2、占有且等待：一个进程本身占有资源（一种或多种），同时还有资源未得到满足，正在等待其他进程释放该资源。  
    3、不可抢占：别人已经占有了某项资源，你不能因为自己也需要该资源，就去把别人的资源抢过来。  
    4、循环等待：存在一个进程链，使得每个进程都占有下一个进程所需的至少一种资源。

1,出现死锁异常后，手工将死锁解开。

2，找出造成死锁的sql：

a，直接看日志：程序中日志做的很详细的话，是能够找到具体哪个sql报的错，操作的哪个表，还有别的模块也操作这个表，线程，并发的程序也会引起。

b,通过oracle的后台v$session表 和 v$sql 的分析 找到。

3,对sql进行优化。

> 悲观锁和乐观锁

悲观锁  
总是假设最坏的情况，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会阻塞直到它拿到锁（共享资源每次只给一个线程使用，其它线程阻塞，用完后再把资源转让给其它线程）。传统的关系型数据库里边就用到了很多这种锁机制，比如行锁，表锁等，读锁，写锁等，都是在做操作之前先上锁。Java中synchronized和ReentrantLock等独占锁就是悲观锁思想的实现  
乐观锁  
总是假设最好的情况，每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号机制和CAS算法实现。乐观锁适用于多读的应用类型，这样可以提高吞吐量，像数据库提供的类似于write\_condition机制，其实都是提供的乐观锁。在Java中java.util.concurrent.atomic包下面的原子变量类就是使用了乐观锁的一种实现方式CAS实现的。

两种锁的使用场景  
从上面对两种锁的介绍，我们知道两种锁各有优缺点，不可认为一种好于另一种，像乐观锁适用于写比较少的情况下（多读场景），即冲突真的很少发生的时候，这样可以省去了锁的开销，加大了系统的整个吞吐量。但如果是多写的情况，一般会经常产生冲突，这就会导致上层应用会不断的进行retry，这样反倒是降低了性能，所以一般多写的场景下用悲观锁就比较合适

> 深入剖析volatile关键字 vɒlətaɪl

一旦一个共享变量（类的成员变量、类的静态成员变量）被volatile修饰之后，那么就具备了两层语义：

保证了不同线程对这个变量进行操作时的可见性，即一个线程修改了某个变量的值，这新值对其他线程来说是立即可见的。

禁止进行指令重排序。

在前面提到volatile关键字能禁止指令重排序，所以volatile能在一定程度上保证有序性

volatile也无法保证对变量的任何操作都是原子性的

可以在定义的 方法上加入  synchronized关键字来实现线程同步或者加入 lock 加锁

public class Test {
        public  int inc = 0;
        
        public synchronized void increase() {
            inc++;
        }
        
        public static void main(String[] args) {
            final Test test = new Test();
            for(int i=0;i<10;i++){
                new Thread(){
                    public void run() {
                        for(int j=0;j<1000;j++)
                            test.increase();
                    };
                }.start();
            }
            
            while(Thread.activeCount()>1)  //保证前面的线程都执行完
                Thread.yield();
            System.out.println(test.inc);
        }
    }

> 说明：

我们使用线程的时候就去创建一个线程，这样实现起来非常简便，但是就会有一个问题：

如果并发的线程数量很多，并且每个线程都是执行一个时间很短的任务就结束了，这样频繁创建线程就会大大降低系统的效率，因为频繁创建线程和销毁线程需要时间。那么有没有一种办法使得线程可以复用，就是执行完一个任务，并不被销毁，而是可以继续执行其他的任务

### 线程池的作用 ###

## **1、线程复用：线程池** ##

假设这里有一个系统，大概每秒需要处理5万条数据，这5万条数据为一个批次，而这没秒发送的5万条数据数据需要经过两个处理过程，第一步是数据存入数据库，第二步是对数据进行其他业务的分析，假设第一步我是用的是普通的JDBC插入数据，为了不影响程序的继续执行，我写了一个线程，让这个子线程不阻塞主线程，继续处理第二步骤的数据，我们知道插入5万条数据大概需要2至3秒的时间，如果每一批次插入数据库的时候，就创建一个线程进行处理，可想而知，由于插入数据库的时间较久，不能很快的处理，这样的话，一段时间之后，系统中就会有很多的这种插入数据的线程，**但是如果使用了线程池这样的话就算再多的数据需要入库，只需要排队等待线程池的线程即可，就不会出现线程池过多而消耗系统资源的情况**

/**
     * 线程池的测试和使用
     */
    public class Test {
        public static void main(String[] args) {
            // 使用ThreadPoolExecutor 创建线程池
    //        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
    //                new ArrayBlockingQueue<Runnable>(5));
           //使用Executor 的静态方法创建线程池
            ExecutorService executor = newFixedThreadPool(5);
            for(int i=0;i<15;i++){
                MyTask myTask = new MyTask(i);
                executor.execute(myTask);
    //            System.out.println("111");
    //            System.out.println("线程池中线程数目："+executor.getPoolSize()+"，队列中等待执行的任务数目："+
    //                    executor.getQueue().size()+"，已执行玩别的任务数目："+executor.getCompletedTaskCount());
            }
            executor.shutdown();
    
            //不使用线程池
            for(int i= 0;i<15;i++){
                MyTask myTask = new MyTask(i);
                Thread thread = new Thread(myTask);
                thread.start();
                System.out.println(thread.getName());
            }
    
        }
    
        public static ExecutorService newFixedThreadPool(int nThreads) {
            return new ThreadPoolExecutor(nThreads, nThreads,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>());
        }
    
        public static ExecutorService newCachedThreadPool() {
            return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                    60L, TimeUnit.SECONDS,
                    new SynchronousQueue<Runnable>());
        }
    }
    
    
    
    
    class MyTask implements Runnable {
        private int taskNum;
    
        public MyTask(int num) {
            this.taskNum = num;
        }
    
        @Override
        public void run() {
            System.out.println("正在执行task "+taskNum);
            try {
                Thread.currentThread().sleep(4000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("task "+taskNum+"执行完毕");
        }
    }

> 线程池最常用的类ThreadPoolExecutor

Executor是一个顶层接口，在它里面只声明了一个方法execute(Runnable)，返回值为void，参数为Runnable类型，从字面意思可以理解，就是用来执行传进去的任务的；

然后ExecutorService接口继承了Executor接口，并声明了一些方法：submit、invokeAll、invokeAny以及shutDown等；

抽象类AbstractExecutorService实现了ExecutorService接口，基本实现了ExecutorService中声明的所有方法；

然后ThreadPoolExecutor继承了类AbstractExecutorService。

在ThreadPoolExecutor类中有几个非常重要的方法：

<table> 
 <tbody> 
  <tr> 
   <td style="width:108px;"> <p>1</p> <p>2</p> <p>3</p> <p>4</p> </td> 
   <td> <p><code>execute()&nbsp;&nbsp;</code>向线程池提交一个任务，交由线程池去执行</p> <p><code>submit()&nbsp;</code>ExecutorService中声明的方法</p> <p><code>&nbsp;</code>这个方法也是用来向线程池提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果，去看submit()&nbsp; 方法的实现，会发现它实际上还是调用的execute()方法，只不过它利用了Future来获取任务执行结果&nbsp;</p> <p><code>shutdown()</code>&nbsp; &nbsp; &nbsp;这两个主要是关闭线程</p> <p><code>shutdownNow()&nbsp;</code></p> </td> 
  </tr> 
 </tbody> 
</table>

## 不过在java doc中，并不提倡我们直接使用ThreadPoolExecutor，而是使用Executors类中提供的几个静态方法来创建线程池： ##

Executors.newCachedThreadPool();        //创建一个缓冲池，缓冲池容量大小为Integer.MAX_VALUE
    Executors.newSingleThreadExecutor();   //创建容量为1的缓冲池
    Executors.newFixedThreadPool(int);    //创建固定容量大小的缓冲池

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

# #

newFixedThreadPool创建的线程池corePoolSize和maximumPoolSize值是相等的，它使用的LinkedBlockingQueue；

newSingleThreadExecutor将corePoolSize和maximumPoolSize都设置为1，也使用的LinkedBlockingQueue；

newCachedThreadPool将corePoolSize设置为0，将maximumPoolSize设置为Integer.MAX\_VALUE，使用的SynchronousQueue，也就是说来了任务就创建线程运行，当线程空闲超过60秒，就销毁线程。

实际中，如果Executors提供的三个静态方法能满足要求，就尽量使用它提供的三个方法，因为自己去手动配置ThreadPoolExecutor的参数有点麻烦，要根据实际任务的类型和数量来进行配置。

# #

## 二.深入剖析线程池实现原理 ##

**1.线程池状态**

**　　2.任务的执行**

**　　3.线程池中的线程初始化**

**　　4.任务缓存队列及排队策略**

**　　5.任务拒绝策略**

**　　6.线程池的关闭**

**　　7.线程池容量的动态调整**

**1.线程池状态**

在ThreadPoolExecutor中定义了一个volatile变量，另外定义了几个static final变量表示线程池的各个状态：

<table> 
 <tbody> 
  <tr> 
   <td> <p>1</p> <p>2</p> <p>3</p> <p>4</p> <p>5</p> </td> 
   <td> <p><code>volatile</code>&nbsp;<code>int</code>&nbsp;<code>runState;</code></p> <p><code>static</code>&nbsp;<code>final</code>&nbsp;<code>int</code>&nbsp;<code>RUNNING&nbsp;&nbsp;&nbsp; =&nbsp;</code><code>0</code><code>;</code></p> <p><code>static</code>&nbsp;<code>final</code>&nbsp;<code>int</code>&nbsp;<code>SHUTDOWN&nbsp;&nbsp; =&nbsp;</code><code>1</code><code>;</code></p> <p><code>static</code>&nbsp;<code>final</code>&nbsp;<code>int</code>&nbsp;<code>STOP&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; =&nbsp;</code><code>2</code><code>;</code></p> <p><code>static</code>&nbsp;<code>final</code>&nbsp;<code>int</code>&nbsp;<code>TERMINATED =&nbsp;</code><code>3</code><code>;</code></p> </td> 
  </tr> 
 </tbody> 
</table>

runState表示当前线程池的状态，它是一个volatile变量用来保证线程之间的可见性；

下面的几个static final变量表示runState可能的几个取值。

当创建线程池后，初始时，线程池处于RUNNING状态；

如果调用了shutdown()方法，则线程池处于SHUTDOWN状态，此时线程池不能够接受新的任务，它会等待所有任务执行完毕；

如果调用了shutdownNow()方法，则线程池处于STOP状态，此时线程池不能接受新的任务，并且会去尝试终止正在执行的任务；

当线程池处于SHUTDOWN或STOP状态，并且所有工作线程已经销毁，任务缓存队列已经清空或执行结束后，线程池被设置为TERMINATED状态

**2.任务的执行**

在了解将任务提交给线程池到任务执行完毕整个过程之前，我们先来看一下ThreadPoolExecutor类中其他的一些比较重要成员变量：

private final BlockingQueue<Runnable> workQueue;              //任务缓存队列，用来存放等待执行的任务
    private final ReentrantLock mainLock = new ReentrantLock();   //线程池的主要状态锁，对线程池状态（比如线程池大小
                                                                  //、runState等）的改变都要使用这个锁
    private final HashSet<Worker> workers = new HashSet<Worker>();  //用来存放工作集
     
    private volatile long  keepAliveTime;    //线程存货时间   
    private volatile boolean allowCoreThreadTimeOut;   //是否允许为核心线程设置存活时间
    private volatile int   corePoolSize;     //核心池的大小（即线程池中的线程数目大于这个参数时，提交的任务会被放进任务缓存队列）
    private volatile int   maximumPoolSize;   //线程池最大能容忍的线程数
     
    private volatile int   poolSize;       //线程池中当前的线程数
     
    private volatile RejectedExecutionHandler handler; //任务拒绝策略
     
    private volatile ThreadFactory threadFactory;   //线程工厂，用来创建线程
     
    private int largestPoolSize;   //用来记录线程池中曾经出现过的最大线程数
     
    private long completedTaskCount;   //用来记录已经执行完毕的任务个数

每个变量的作用都已经标明出来了，这里要重点解释一下corePoolSize、maximumPoolSize、largestPoolSize三个变量。

corePoolSize在很多地方被翻译成核心池大小，其实我的理解这个就是线程池的大小。举个简单的例子：

假如有一个工厂，工厂里面有10个工人，每个工人同时只能做一件任务。

因此只要当10个工人中有工人是空闲的，来了任务就分配给空闲的工人做；

当10个工人都有任务在做时，如果还来了任务，就把任务进行排队等待；

如果说新任务数目增长的速度远远大于工人做任务的速度，那么此时工厂主管可能会想补救措施，比如重新招4个临时工人进来；

然后就将任务也分配给这4个临时工人做；

如果说着14个工人做任务的速度还是不够，此时工厂主管可能就要考虑不再接收新的任务或者抛弃前面的一些任务了。

当这14个工人当中有人空闲时，而新任务增长的速度又比较缓慢，工厂主管可能就考虑辞掉4个临时工了，只保持原来的10个工人，毕竟请额外的工人是要花钱的。

这个例子中的corePoolSize就是10，而maximumPoolSize就是14（10+4）。

也就是说corePoolSize就是线程池大小，maximumPoolSize在我看来是线程池的一种补救措施，即任务量突然过大时的一种补救措施。

不过为了方便理解，在本文后面还是将corePoolSize翻译成核心池大小。

largestPoolSize只是一个用来起记录作用的变量，用来记录线程池中曾经有过的最大线程数目，跟线程池的容量没有任何关系。

execute()方法的实现原理即可

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
        }
    }

2.当调用 execute() 方法添加一个任务时，线程池会做如下判断：

*  如果正在运行的线程数量小于 corePoolSize，那么马上创建线程运行这个任务；
 *  如果正在运行的线程数量大于或等于 corePoolSize，那么将这个任务放入队列；
 *  如果这时候队列满了，而且正在运行的线程数量小于 maximumPoolSize，那么还是要创建非核心线程立刻运行这个任务；
 *  如果队列满了，而且正在运行的线程数量大于或等于 maximumPoolSize，那么线程池会抛出异常RejectExecutionException。

**3.线程池中的线程初始化**

默认情况下，创建线程池之后，线程池中是没有线程的，需要提交任务之后才会创建线程。

在实际中如果需要线程池创建之后立即创建线程，可以通过以下两个方法办到：

*  prestartCoreThread()：初始化一个核心线程；
 *  prestartAllCoreThreads()：初始化所有核心线程

**6.线程池的关闭**

ThreadPoolExecutor提供了两个方法，用于线程池的关闭，分别是shutdown()和shutdownNow()，其中：

*  shutdown()：不会立即终止线程池，而是要等所有任务缓存队列中的任务都执行完后才终止，但再也不会接受新的任务
 *  shutdownNow()：立即终止线程池，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务