JUC学习 - 线程池1

JUC学习 - 线程池1 一、线程池 1、什么是线程池

大家用jdbc *** 作过数据库应该知道， *** 作数据库需要和数据库建立连接，拿到连接之后才能 *** 作数据库，用完之后销毁。数据库连接的创建和销毁其实是比较耗时的，真正和业务相关的 *** 作耗时是比较短的。每个数据库 *** 作之前都需要创建连接，为了提升系统性能，后来出现了数据库连接池，系统启动的时候，先创建很多连接放在池子里面，使用的时候，直接从连接池中获取一个，使用完毕之后返回到池子里面，继续给其他需要者使用，这其中就省去创建连接的时间，从而提升了系统整体的性能。

线程池和数据库连接池的原理也差不多，创建线程去处理业务，可能创建线程的时间比处理业务的时间还长一些，如果系统能够提前为我们创建好线程，我们需要的时候直接拿来使用，用完之后不是直接将其关闭，而是将其返回到线程池中，给其他需要这使用，这样直接节省了创建和销毁的时间，提升了系统的性能。

简单的说，在使用了线程池之后，创建线程变成了从线程池中获取一个空闲的线程，然后使用，关闭线程变成了将线程归还到线程池。

2、线程池实现原理

当向线程池提交一个任务之后，线程池的处理流程如下：

1. 判断是否达到核心线程数，若未达到，则直接创建新的线程处理当前传入的任务，否则进入下个流程
2. 线程池中的工作队列是否已满，若未满，则将任务丢入工作队列中先存着等待处理，否则进入下个流程
3. 是否达到最大线程数，若未达到，则创建新的线程处理当前传入的任务，否则交给线程池中的饱和策略进行处理。

流程如下图：

举个例子，加深理解：

咱们作为开发者，上面都有开发主管，主管下面带领几个小弟干活，CTO给主管授权说，你可以招聘5个小弟干活，新来任务，如果小弟还不到5个，立即去招聘一个来干这个新来的任务，当5个小弟都招来了，再来任务之后，将任务记录到一个表格中，表格中最多记录100个，小弟们会主动去表格中获取任务执行，如果5个小弟都在干活，并且表格中也记录满了，那你可以将小弟扩充到20个，如果20个小弟都在干活，并且存放任务的表也满了，产品经理再来任务后，是直接拒绝，还是让产品自己干，这个由你自己决定，小弟们都尽心尽力在干活，任务都被处理完了，突然公司业绩下滑，几个员工没事干，打酱油，为了节约成本，CTO让主管把小弟控制到5人，其他15个人直接被干掉了。所以作为小弟们，别让自己闲着，多干活。

原理：先找几个人干活，大家都忙于干活，任务太多可以排期，排期的任务太多了，再招一些人来干活，最后干活的和排期都达到上层领导要求的上限了，那需要采取一些其他策略进行处理了。对于长时间不干活的人，考虑将其开掉，节约资源和成本。

2、java中的线程池

jdk中提供了线程池的具体实现，实现类是：java.util.concurrent.ThreadPoolExecutor，主要构造方法：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)

• corePoolSize：核心线程大小，当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使有其他空闲线程可以处理任务也会创建新线程，等到工作的线程数大于核心线程数时就不会在创建了。如果调用了线程池的prestartAllCoreThreads方法，线程池会提前把核心线程都创造好，并启动。
• maximumPoolSize：线程池允许创建的最大线程数。如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。如果我们使用了无界队列，那么所有的任务会加入队列，这个参数就没有什么效果了。
• keepAliveTime：线程池的工作线程空闲后，保持存活的时间。如果没有任务处理了，有些线程会空闲，空闲的时间超过了这个值，会被回收掉。如果任务很多，并且每个任务的执行时间比较短，避免线程重复创建和回收，可以调大这个时间，提高线程的利用率。
• unit：keepAliveTIme的时间单位，可以选择的单位有天、小时、分钟、毫秒、微妙、千分之一毫秒和纳秒。类型是一个枚举java.util.concurrent.TimeUnit，这个枚举也经常使用，有兴趣的可以看一下其源码。
• workQueue：工作队列，用于缓存待处理任务的阻塞队列，常见的有4种，本文后面有介绍。
• threadFactory：线程池中创建线程的工厂，可以通过线程工厂给每个创建出来的线程设置更有意义的名字。
• handler：饱和策略，当线程池无法处理新来的任务了，那么需要提供一种策略处理提交的新任务，默认有4种策略，文章后面会提到。

调用线程池的execute方法处理任务，执行execute方法的过程：

1. 判断线程池中运行的线程数是否小于corepoolsize，是：则创建新的线程来处理任务，否：执行下一步。
2. 试图将任务添加到workQueue指定的队列中，如果无法添加到队列，进入下一步。
3. 判断线程池中运行的线程数是否小于maximumPoolSize，是：则新增线程处理当前传入的任务，否：将任务传递给handler对象rejectedExecution方法处理。

线程池的使用步骤：

1. 调用构造方法创建线程池
2. 调用线程池的方法处理任务
3. 关闭线程池

3、线程池使用的简单示例

上一个简单的示例，如下：

public class ThreadPoolTest {
    static ThreadPoolExecutor executor 
           = new ThreadPoolExecutor(3,   // 核心线程数
                                    5,	 // 最大线程数
                                    10,	 // 存活时间
                                    TimeUnit.SECONDS,   // 时间单位 s
                                    new ArrayBlockingQueue(10),   // 等待队列 10
                                    Executors.defaultThreadFactory(),  // 线程创建工厂
                                    new ThreadPoolExecutor.AbortPolicy());   // 饱和策略
    
    public static void main(String[] args) {
        for (int i = 0; i < 10; i++) {
            int j = i;
            String taskName = "任务" + j;
            
            executor.execute(() -> {
                //模拟任务内部处理耗时
                try {
                    TimeUnit.SECONDS.sleep(j);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + taskName + "处理完毕");
            });
        }
        //关闭线程池
        executor.shutdown();
    }
}

运行上面的代码，输出结果：

pool-1-thread-1任务0处理完毕
pool-1-thread-2任务1处理完毕
pool-1-thread-3任务2处理完毕
pool-1-thread-1任务3处理完毕
pool-1-thread-2任务4处理完毕
pool-1-thread-3任务5处理完毕
pool-1-thread-1任务6处理完毕
pool-1-thread-2任务7处理完毕
pool-1-thread-3任务8处理完毕
pool-1-thread-1任务9处理完毕

4、线程池中常见5种工作队列

任务太多的时候，工作队列用于暂时缓存待处理的任务，jdk中常见的5种阻塞队列：

• ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按照先进先出原则对元素进行排序。
• linkedBlockingQueue：是一个基于链表结构的阻塞队列，此队列按照先进先出排序元素，吞吐量通常要高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool使用了这个队列。
• SynchronousQueue ：一个不存储元素的阻塞队列，每个插入 *** 作必须等到另外一个线程调用移除 *** 作，否则插入 *** 作一直处理阻塞状态，吞吐量通常要高于linkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用这个队列。
• PriorityBlockingQueue：优先级队列，进入队列的元素按照优先级会进行排序。

5、SynchronousQueue队列的线程池

public class SynchronousQueueTest {
    
    public static void main(String[] args) {
        ExecutorService executor = Executors.newCachedThreadPool();
        for (int i = 0; i < 50; i++) {
            int j = i;
            String taskName = "任务" + j;
            
            executor.execute(() -> {
                System.out.println(Thread.currentThread().getName() + "处理" + taskName);
                //模拟任务内部处理耗时 1s
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            
        }
        //关闭线程池
        executor.shutdown();
    }
    
}

运行上面的代码，输出结果：

pool-1-thread-1处理任务0
pool-1-thread-2处理任务1
pool-1-thread-3处理任务2
pool-1-thread-4处理任务3
pool-1-thread-5处理任务4
pool-1-thread-6处理任务5
pool-1-thread-7处理任务6
pool-1-thread-8处理任务7
pool-1-thread-9处理任务8
pool-1-thread-10处理任务9
pool-1-thread-12处理任务11
pool-1-thread-45处理任务44
pool-1-thread-15处理任务14
pool-1-thread-14处理任务13
pool-1-thread-13处理任务12
pool-1-thread-16处理任务15
pool-1-thread-17处理任务16
pool-1-thread-18处理任务17
pool-1-thread-19处理任务18
pool-1-thread-20处理任务19
pool-1-thread-21处理任务20
pool-1-thread-22处理任务21
pool-1-thread-23处理任务22
pool-1-thread-24处理任务23
pool-1-thread-26处理任务25
pool-1-thread-25处理任务24
pool-1-thread-27处理任务26
pool-1-thread-28处理任务27
pool-1-thread-29处理任务28
pool-1-thread-30处理任务29
pool-1-thread-31处理任务30
pool-1-thread-32处理任务31
pool-1-thread-33处理任务32
pool-1-thread-34处理任务33
pool-1-thread-35处理任务34
pool-1-thread-36处理任务35
pool-1-thread-37处理任务36
pool-1-thread-38处理任务37
pool-1-thread-39处理任务38
pool-1-thread-40处理任务39
pool-1-thread-41处理任务40
pool-1-thread-43处理任务42
pool-1-thread-48处理任务47
pool-1-thread-49处理任务48
pool-1-thread-44处理任务43
pool-1-thread-50处理任务49
pool-1-thread-42处理任务41
pool-1-thread-11处理任务10
pool-1-thread-47处理任务46
pool-1-thread-46处理任务45

代码中使用Executors.newCachedThreadPool()创建线程池，看一下的源码：

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0,    // 核心线程数 0
                                  Integer.MAX_VALUE,  // 最大线程数 
                                  60L,  // 存活时间
                                  TimeUnit.SECONDS,   // 时间单位
                                  new SynchronousQueue());   // 等待队列
}

从输出中可以看出，系统创建了50个线程处理任务，代码中使用了SynchronousQueue同步队列，这种队列比较特殊，放入元素必须要有另外一个线程去获取这个元素，否则放入元素会失败或者一直阻塞在那里直到有线程取走，示例中任务处理休眠了指定的时间，导致已创建的工作线程都忙于处理任务，所以新来任务之后，将任务丢入同步队列会失败，丢入队列失败之后，会尝试新建线程处理任务。

使用上面的方式创建线程池需要注意，如果需要处理的任务比较耗时，会导致新来的任务都会创建新的线程进行处理，可能会导致创建非常多的线程，最终耗尽系统资源，触发OOM。

6、PriorityBlockingQueue优先级队列的线程池

public class PriorityBlockingQueueTest {
    
    static class Task implements Runnable, Comparable {
        private int i;
        private String name;
        
        public Task(int i, String name) {
            this.i = i;
            this.name = name;
        }
        
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + "处理" + this.name);
        }
        
        @Override
        public int compareTo(Task o) {
            return Integer.compare(o.i, this.i);
        }
    }
    
    public static void main(String[] args) {
        ExecutorService executor = new ThreadPoolExecutor(1, 1,
                60L, TimeUnit.SECONDS,
                new PriorityBlockingQueue());
        
        for (int i = 0; i < 10; i++) {
            String taskName = "任务" + i;
            executor.execute(new Task(i, taskName));
        }
        
        for (int i = 100; i >= 90; i--) {
            String taskName = "任务" + i;
            executor.execute(new Task(i, taskName));
        }
        
        executor.shutdown();
    }
    
}

运行上面的代码，输出结果：

pool-1-thread-1处理任务0
pool-1-thread-1处理任务100
pool-1-thread-1处理任务99
pool-1-thread-1处理任务98
pool-1-thread-1处理任务97
pool-1-thread-1处理任务96
pool-1-thread-1处理任务95
pool-1-thread-1处理任务94
pool-1-thread-1处理任务93
pool-1-thread-1处理任务92
pool-1-thread-1处理任务91
pool-1-thread-1处理任务90
pool-1-thread-1处理任务9
pool-1-thread-1处理任务8
pool-1-thread-1处理任务7
pool-1-thread-1处理任务6
pool-1-thread-1处理任务5
pool-1-thread-1处理任务4
pool-1-thread-1处理任务3
pool-1-thread-1处理任务2
pool-1-thread-1处理任务1

输出中，除了第一个任务，其他任务按照优先级高低按顺序处理。

原因在于：创建线程池的时候使用了优先级队列，进入队列中的任务会进行排序，任务的先后顺序由Task中的i变量决定。向PriorityBlockingQueue加入元素的时候，内部会调用代码中Task的compareTo方法决定元素的先后顺序。

7、自定义创建线程的工厂

给线程池中线程起一个有意义的名字，在系统出现问题的时候，通过线程堆栈信息可以更容易发现系统中问题所在。

自定义创建工厂需要实现java.util.concurrent.ThreadFactory接口中的Thread newThread(Runnable r)方法，参数为传入的任务，需要返回一个工作线程。

示例代码：

public class CustomizeThreadFactory {

    static AtomicInteger threadNum = new AtomicInteger(1);

    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 5,
                60L, TimeUnit.SECONDS,
                new ArrayBlockingQueue(10), new ThreadFactory() {
                            @Override
                            public Thread newThread(Runnable r) {
                                Thread thread = new Thread(r);
                                thread.setName("自定义线程-" + threadNum.getAndIncrement());
                                return thread;
                            }
        });

        for (int i = 0; i < 5; i++) {
            String taskName = "任务-" + i;
            executor.execute(() -> {
                System.out.println(Thread.currentThread().getName() + "处理" + taskName);
            });
        }

        executor.shutdown();
    }

}

运行上面的代码，输出结果：

自定义线程-2处理任务-1
自定义线程-5处理任务-4
自定义线程-4处理任务-3
自定义线程-1处理任务-0
自定义线程-3处理任务-2

代码中在任务中输出了当前线程的名称，可以看到是我们自定义的名称。

通过jstack查看线程的堆栈信息，也可以看到我们自定义的名称，我们可以将代码中executor.shutdown();先给注释掉让程序先不退出，然后通过jstack查看，如下：

"自定义线程-5" #15 prio=5 os_prio=0 tid=0x00000000194fb000 nid=0x4ff0 waiting on condition [0x000000001a26f000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000d6090520> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
        at java.util.concurrent.ArrayBlockingQueue.take(ArrayBlockingQueue.java:403)
        at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

"自定义线程-4" #14 prio=5 os_prio=0 tid=0x00000000194fa800 nid=0x3450 waiting on condition [0x000000001a16f000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000d6090520> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
        at java.util.concurrent.ArrayBlockingQueue.take(ArrayBlockingQueue.java:403)
        at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

"自定义线程-3" #13 prio=5 os_prio=0 tid=0x00000000194f5800 nid=0x4808 waiting on condition [0x000000001a06f000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000d6090520> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
        at java.util.concurrent.ArrayBlockingQueue.take(ArrayBlockingQueue.java:403)
        at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

"自定义线程-2" #12 prio=5 os_prio=0 tid=0x00000000194f5000 nid=0x1c10 waiting on condition [0x0000000019f6e000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000d6090520> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
        at java.util.concurrent.ArrayBlockingQueue.take(ArrayBlockingQueue.java:403)
        at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

"自定义线程-1" #11 prio=5 os_prio=0 tid=0x00000000194f4000 nid=0x20bc waiting on condition [0x0000000019e6e000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000d6090520> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
        at java.util.concurrent.ArrayBlockingQueue.take(ArrayBlockingQueue.java:403)
        at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

文章参考：http://www.itsoku.com/ 博主觉得这个文章的内容挺不错的，感兴趣的可以去了解一下。