多线程

多线程，是对资源的利用方式，以期望使用更少的资源，做更多的事情，提升 CPU、IO等资源的利用率。
java 实现多线程的方式：runnable，thread
thread run、start、join 的几个方法的区别：

  1. run 当前线程运行，串行，不能起到多线程的效果
  3. start 另起线程执行，并行
  5. join 等待线程执行完，阻止当前线程退出，多与 start 方法一起使用

代码

  1. package com.xiaohui.demo.webfluxclient.thread;
  3. import java.util.concurrent.TimeUnit;
  5. /**
  6.  * <p>
  7.  * 测试多线程
  8.  * </p>
  9.  *
  10.  * @author xiaohui
  11.  * @email xuxiaohuimail@163.com
  12.  * @since 2021/6/22
  13.  */
  14. public class TestThread {
  16.     public static void main(String[] args) throws InterruptedException {
  17.         System.out.println(">>>>>>>>>>>>>>>>>>>> start " + Thread.currentThread().getName());
  18.         // run 方法，直接被当前线程调用业务代码
  19.         MyThread mythead001 = new MyThread("mythead001");
  20.         mythead001.run();
  21.         // start 方法，另起线程执行当前业务代码
  22.         MyThread mythead002 = new MyThread("mythead002");
  23.         mythead002.start();
  24.         // join 将当前的线程执行完成，再执行后面的业务代码（start会直接继续执行后续代码，不会等待返回），join 一般与start 一并使用
  25.         MyThread mythead003 = new MyThread("mythead003");
  26.         mythead003.start();
  27.         mythead003.join();
  28.         System.out.println(">>>>>>>>>>>>>>>>>>>> end " + Thread.currentThread().getName());
  29.     }
  31.     static class MyThread extends Thread {
  32.         public MyThread(String myName) {
  33.             this.myName = myName;
  34.         }
  36.         public String getMyName() {
  37.             return myName;
  38.         }
  40.         public void setMyName(String myName) {
  41.             this.myName = myName;
  42.         }
  44.         private String myName;
  46.         @Override
  47.         public void run() {
  48.             try {
  49.                 System.out.println(">>>>>>>>>>>>>>>>>>> 进入mythread 。。。threadName:" + Thread.currentThread().getName() + ",参数：" + getMyName());
  50.                 TimeUnit.SECONDS.sleep(10);
  51.                 System.out.println(">>>>>>>>>>>>>>>>>>> 退出mythread 。。。threadName:" + Thread.currentThread().getName() + ",参数：" + getMyName());
  52.             } catch (InterruptedException e) {
  53.                 e.printStackTrace();
  54.             }
  55.         }
  56.     }
  58.     static class MyRunnable implements Runnable {
  59.         public MyRunnable(String myName) {
  60.             this.myName = myName;
  61.         }
  63.         public String getMyName() {
  64.             return myName;
  65.         }
  67.         public void setMyName(String myName) {
  68.             this.myName = myName;
  69.         }
  71.         private String myName;
  74.         @Override
  75.         public void run() {
  76.             try {
  77.                 System.out.println(">>>>>>>>>>>>>>>>>>> 进入MyRunnable 。。。threadName:" + Thread.currentThread().getName() + ",参数：" + getMyName());
  78.                 TimeUnit.SECONDS.sleep(10);
  79.                 System.out.println(">>>>>>>>>>>>>>>>>>> 退出MyRunnable 。。。threadName:" + Thread.currentThread().getName() + ",参数：" + getMyName());
  80.             } catch (InterruptedException e) {
  81.                 e.printStackTrace();
  82.             }
  83.         }
  84.     }
  85. }

线程池

threadpool，

有了多线程，为什么还需要线程池？

创建线程消耗时间多，资源多，线程越多，切换线程消耗时间长。线程池可以复用线程。
image.png

代码测试：

  1. package com.xiaohui.demo.webfluxclient.thread;
  3. import org.junit.jupiter.api.Test;
  5. import java.util.ArrayList;
  6. import java.util.List;
  7. import java.util.Random;
  8. import java.util.concurrent.CopyOnWriteArrayList;
  9. import java.util.concurrent.ExecutorService;
  10. import java.util.concurrent.Executors;
  11. import java.util.concurrent.TimeUnit;
  13. /**
  14.  * <p>
  15.  * 测试线程池
  16.  * </p>
  17.  *
  18.  * @author xiaohui
  19.  * @email xuxiaohuimail@163.com
  20.  * @since 2021/6/23
  21.  */
  22. public class TestThreadPool {
  24.     /**
  25.      * 测试非线程池下的代码
  26.      *
  27.      * @throws InterruptedException
  28.      */
  29.     @Test
  30.     public void testNonThreadPool() throws InterruptedException {
  31.         System.out.println(">>>>>>>>>>>>>>>>>>>>> testNonThreadPool");
  32.         long start = System.currentTimeMillis();
  33.         final List<Integer> list = new ArrayList<>();
  34.         for (int i = 0; i < 100000; i++) {
  35.             Thread thread = new Thread(() -> {
  36.                 list.add(new Random().nextInt());
  37.             });
  38.             thread.start();
  39.             thread.join();
  40.         }
  41.         System.out.println(String.format("list size %d", list.size()));
  42.         System.out.println(String.format("cost time %d", System.currentTimeMillis() - start));
  44.     }
  46.     @Test
  47.     public void testThreadPool() throws InterruptedException {
  48.         System.out.println(">>>>>>>>>>>>>>>>>>>>> testThreadPool");
  49.         long start = System.currentTimeMillis();
  50.         final List<Integer> list = new ArrayList<>();
  51.         ExecutorService executorService = Executors.newFixedThreadPool(1000);
  52.         for (int i = 0; i < 100000; i++) {
  53.             /*
  54.             >>>>>>>>>>>>>>>>>>>>> testThreadPool
  55.             list size 99931
  56.             cost time 5705
  57.              */
  58.             // 使用 submit， 执行结果 list size 小于 100000，ArrayList 非线程安全
  59.             executorService.submit(() -> {
  60.                 list.add(new Random().nextInt());
  61.             });
  62.         }
  63.         executorService.shutdown();
  64.         executorService.awaitTermination(1, TimeUnit.HOURS);
  66.         System.out.println(String.format("list size %d", list.size()));
  67.         System.out.println(String.format("cost time %d", System.currentTimeMillis() - start));
  68.     }
  70.     @Test
  71.     public void testThreadPool2() throws InterruptedException {
  72.         System.out.println(">>>>>>>>>>>>>>>>>>>>> testThreadPool2");
  73.         long start = System.currentTimeMillis();
  74.         final List<Integer> list = new ArrayList<>();
  75.         ExecutorService executorService = Executors.newFixedThreadPool(1000);
  76.         for (int i = 0; i < 100000; i++) {
  77.             /*
  78.             >>>>>>>>>>>>>>>>>>>>> testThreadPool2
  79.             list size 99985
  80.             cost time 992
  81.              */
  82.             // 使用 execute， 执行结果 list size 小于 100000，ArrayList 非线程安全
  83.             executorService.execute(() -> {
  84.                 list.add(new Random().nextInt());
  85.             });
  86.         }
  87.         executorService.shutdown();
  88.         executorService.awaitTermination(1, TimeUnit.HOURS);
  90.         System.out.println(String.format("list size %d", list.size()));
  91.         System.out.println(String.format("cost time %d", System.currentTimeMillis() - start));
  92.     }
  94.     @Test
  95.     public void testThreadPool3() throws InterruptedException {
  96.         System.out.println(">>>>>>>>>>>>>>>>>>>>> testThreadPool3");
  97.         long start = System.currentTimeMillis();
  98. //        final List<Integer> list = new ArrayList<>();
  99.         // 采用线程安全的 list
  100.         CopyOnWriteArrayList<Integer> list = new CopyOnWriteArrayList<>();
  101.         ExecutorService executorService = Executors.newFixedThreadPool(1000);
  102.         for (int i = 0; i < 100000; i++) {
  103.             /*
  104.             >>>>>>>>>>>>>>>>>>>>> testThreadPool3
  105.             list size 100000
  106.             cost time 13683
  107.              */
  108.             // 使用 execute， 执行结果 list size 等于 100000，CopyOnWriteArrayList 非线程安全
  109.             executorService.execute(() -> {
  110.                 list.add(new Random().nextInt());
  111.             });
  112.         }
  113.         executorService.shutdown();
  114.         executorService.awaitTermination(1, TimeUnit.HOURS);
  116.         System.out.println(String.format("list size %d", list.size()));
  117.         System.out.println(String.format("cost time %d", System.currentTimeMillis() - start));
  118.     }
  120.     /**
  121.      * 采用 Executors.newSingleThreadExecutor() 线程池，不存在线程安全问题
  122.      *
  123.      * @throws InterruptedException
  124.      */
  125.     @Test
  126.     public void testThreadPool4() throws InterruptedException {
  127.         System.out.println(">>>>>>>>>>>>>>>>>>>>> testThreadPool4");
  128.         long start = System.currentTimeMillis();
  129.         final List<Integer> list = new ArrayList<>();
  130.         // 采用单一线程池执行
  131.         ExecutorService executorService = Executors.newSingleThreadExecutor();
  132.         for (int i = 0; i < 100000; i++) {
  133.             /*
  134.             >>>>>>>>>>>>>>>>>>>>> testThreadPool4
  135.             list size 100000
  136.             cost time 150
  137.              */
  138.             // 使用 execute， 执行结果 list size 等于 100000，单一线程，不存在是否多线程的安全问题
  139.             executorService.execute(() -> {
  140.                 list.add(new Random().nextInt());
  141.             });
  142.         }
  143.         executorService.shutdown();
  144.         executorService.awaitTermination(1, TimeUnit.HOURS);
  146.         System.out.println(String.format("list size %d", list.size()));
  147.         System.out.println(String.format("cost time %d", System.currentTimeMillis() - start));
  148.     }
  151. }

为什么不推荐使用 Executors 工具类创建线程池

容易导致 CPU 使用率高，比如 Executors.newCacheThreadPool()，没有缓存的线程时，一个劲儿创建线程；
容易导致 OOM 内存溢出，比如 Executors.newFixedThreadPool(10)、Executors.newSingleThreadExecutor()，队列过大，耗尽内存；

它的原理（源码理解）

image.png
image.png
image.png
image.png
image.png
image.png
image.png

如何使用

代码。使用工具类 Executors

  1. package com.xiaohui.demo.webfluxclient.thread;
  3. import org.junit.jupiter.api.Test;
  5. import java.util.concurrent.ExecutorService;
  6. import java.util.concurrent.Executors;
  7. import java.util.concurrent.TimeUnit;
  9. /**
  10.  * <p>
  11.  * 测试线程池2
  12.  * </p>
  13.  *
  14.  * @author xiaohui
  15.  * @email xuxiaohuimail@163.com
  16.  * @since 2021/6/23
  17.  */
  18. public class TestTreadPool_2 {
  20.     @Test
  21.     public void testMultiPool() throws InterruptedException {
  22.         ExecutorService executorService1 = Executors.newSingleThreadExecutor();
  23.         ExecutorService executorService2 = Executors.newFixedThreadPool(100);
  24.         ExecutorService executorService3 = Executors.newCachedThreadPool();
  27.         System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
  28.         System.out.println("Executors.newSingleThreadExecutor() cost time:" + costTime(executorService1));
  29.         System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
  30.         System.out.println("Executors.Executors.newFixedThreadPool(100) cost time:" + costTime(executorService2));
  31.         System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
  32.         System.out.println("Executors.newCachedThreadPool() cost time:" + costTime(executorService3));
  33.         // 結果
  34.         /*
  35.         >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
  36.         curr thread name: pool-1-thread-1, i 为 0
  37.         curr thread name: pool-1-thread-1, i 为 1
  38.         curr thread name: pool-1-thread-1, i 为 2
  39.         curr thread name: pool-1-thread-1, i 为 3
  40.         curr thread name: pool-1-thread-1, i 为 4
  41.         Executors.newSingleThreadExecutor() cost time:10021
  42.         >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
  43.         curr thread name: pool-2-thread-1, i 为 0
  44.         curr thread name: pool-2-thread-2, i 为 1
  45.         curr thread name: pool-2-thread-3, i 为 2
  46.         curr thread name: pool-2-thread-4, i 为 3
  47.         curr thread name: pool-2-thread-5, i 为 4
  48.         Executors.Executors.newFixedThreadPool(100) cost time:2000
  49.         >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
  50.         curr thread name: pool-3-thread-1, i 为 0
  51.         curr thread name: pool-3-thread-5, i 为 4
  52.         curr thread name: pool-3-thread-4, i 为 3
  53.         curr thread name: pool-3-thread-2, i 为 1
  54.         curr thread name: pool-3-thread-3, i 为 2
  55.         Executors.newCachedThreadPool() cost time:2017
  56.          */
  57.     }
  60.     private Long costTime(ExecutorService executorService) throws InterruptedException {
  61.         long start = System.currentTimeMillis();
  62.         for (int i = 0; i < 5; i++) {
  63.             final int a = i;
  64.             executorService.execute(() -> {
  65.                 try {
  66.                     TimeUnit.SECONDS.sleep(2);
  67.                     System.out.println(String.format("curr thread name: %s, i 为 %d ", Thread.currentThread().getName(), a));
  68.                 } catch (InterruptedException e) {
  69.                     e.printStackTrace();
  70.                 }
  71.             });
  72.         }
  73.         executorService.shutdown();
  74.         executorService.awaitTermination(1, TimeUnit.HOURS);
  75.         return System.currentTimeMillis() - start;
  76.     }
  77. }

线程调优

术语介绍

线程与进程

进程，资源分配的最小单位；
线程，CPU 调度的最小单位；也叫轻量级进程 LWP（Light Weight Process）
一个进程拥有多个线程，一个线程属于某一个进程；

ULT 与 KLT 线程模型

用户级线程，user level thread，用户自己创建管理，操作系统（kernel）对线程的存在与否无感知。
image.png
内核级先吃，kernel level thread，依赖内核系统，与内核系统线程一对一管理
image.png
image.png
image.png
程序计数器，pc 寄存器，CPU 时间片等概念

阻塞队列

FIFO
在任意时刻（不管是否高并发），永远只有一个线程能够进行队列的入队或者出队操作！
线程安全的队列！
（实际）有界！（不管指不指定大小，最终取决于宿主操作系统）；
队列满，只能进行出队操作，所有的入队操作必须等待，也就是被阻塞；
队列空，只能进行入队操作，所有的出队操作必须等待，也就是被阻塞；

提交优先级与执行优先级

线程池的提交优先级与执行优先级，都是代码实现，限定了它的表现形式。所以，查看源码吧~~~
任务提交数量：
提交优先：
小于 coreSize 线程，创建线程，执行任务
大于 coreSize 线程，小于 blockQueue size，提交到阻塞队列中
大于 blockQueue size，小于 maxSize 线程， 创建线程，执行任务
大于 blockQueue size，大于 maxSize 线程，采用相关的策略来进行处理

执行优先级：
coreSize 的 thread 先行
maxSize 的在后（任务未能填满阻塞队列时，maxSize 的线程不会创建，即任务在会在 coreSize 线程上执行）；

线程中断信号，安全退出