四种创建线程的方式：

  1. 继承Thread类（是一个Runnable接口的实现类）
  2. 实现Runnable接口（常用，使用lambda表达式扔到Thread类中执行）
  3. 实现callable接口（有返回值和范型，使用FutureTask类，再扔到Thread类中执行）

  4. 使用Executors来创建线程池再创建线程（使用ThreadPool）

    lambda与内部类

    ```java public class TestLambda { // 2.静态内部类 static class Like2 implements Like{

    1.  @Override
    2.  public void lambda() {
    3.      System.out.println("l2");
    4.  }

    }

    public static void main(String[] args) {

    1.  // 1.传统方法
    2.  Like l1 = new ILike();
    3.  l1.lambda();
    5.  // 2.静态内部类
    6.  Like2 l2 = new Like2();
    7.  l2.lambda();
    9.  // 3.局部内部类
    10.  class Like3 implements Like{
    12.      @Override
    13.      public void lambda() {
    14.          System.out.println("l3");
    15.      }
    16.  }
    18.  Like3 l3 = new Like3();
    19.  System.out.println("l3");
    21.  // 4. 匿名内部类
    22.  Like l4 = new Like(){
    23.      @Override
    24.      public void lambda() {
    25.          System.out.println("l4");
    26.      }
    27.  };
    28.  l4.lambda();
    30.  // 5. lambda
    31.  Like l5 = ()->System.out.println("l5");
    33.  l5.lambda();

    }

} // 定义一个接口 interface Like{ void lambda(); }

// 定义接口实现类 class ILike implements Like{

  1. @Override
  2. public void lambda() {
  3.     System.out.println("l1");
  4. }

}

  2. <a name="GJUMu"></a>
  3. ### Runnable + Synchronized锁 + wait、notifyAll线程通信方式
  5. - Synchronized锁,普通方法锁的是实例对象，静态方法锁的是class类模板。
  6. - Synchronized锁有两种使用方式
  7. 1. 锁方法
  9. public Synchronized void add(){}
  11. 2. 锁对象  (一般锁有共享访问属性的对象)
  13. Synchronized(obj){}
  15. - 生产者消费者模式：判断 -> 不满足等待（wait） -> 满足就执行并唤醒（notifyAll）
  16. ```java
  17. /**
  18.  * synchronized方法版
  19.  * 使用OOP类封装属性和同步方法，再去调用Thread线程，传入实例对象的方法开启多线程
  20.  */
  21. public class TestJucSyn {
  22.     public static void main(String[] args) {
  23.         // 实例一个需要多线程操作的对象
  24.         Fun fun = new Fun();
  26.         // 使用lambda表达式简化
  27.         // 将对象需要多线程操作的方法用lambda重写run()方法 
  28.         // 再传给Thread接口开启多线程操作
  29.         // 模板 new Thread(()->{}, "Name").start();
  30.         // 开启4个线程
  31.         new Thread(()->{
  32.             // @Override
  33.             // public void run(){
  34.             for (int i = 0; i < 10; i++) {
  35.                 try {
  36.                     fun.increament();
  37.                 } catch (InterruptedException e) {
  38.                     e.printStackTrace();
  39.                 }
  40.             }
  41.         }, "A").start();
  42.         new Thread(()->{
  43.             for (int i = 0; i < 10; i++) {
  44.                 try {
  45.                     fun.decreament();
  46.                 } catch (InterruptedException e) {
  47.                     e.printStackTrace();
  48.                 }
  49.             }
  50.         }, "B").start();
  52.         new Thread(()->{
  53.             for (int i = 0; i < 10; i++) {
  54.                 try {
  55.                     fun.increament();
  56.                 } catch (InterruptedException e) {
  57.                     e.printStackTrace();
  58.                 }
  59.             }
  60.         }, "C").start();
  61.         new Thread(()->{
  62.             for (int i = 0; i < 10; i++) {
  63.                 try {
  64.                     fun.decreament();
  65.                 } catch (InterruptedException e) {
  66.                     e.printStackTrace();
  67.                 }
  68.             }
  69.         }, "D").start();
  70.     }
  71. }
  74. class Fun {
  75.     private int num = 0;
  77.     public synchronized void increament() throws InterruptedException {
  78.         // 使用while 防止多线程唤醒失败
  79.         while (num == 1) {
  80.             this.wait();
  81.         }
  82.         num++;
  83.         System.out.println(Thread.currentThread().getName() +"=>"+ num);
  84.         this.notifyAll();
  85.     }
  87.     public synchronized void decreament() throws InterruptedException {
  88.         while (num == 0) {
  89.             this.wait();
  90.         }
  91.         num--;
  92.         System.out.println(Thread.currentThread().getName() +"=>"+ num);
  93.         this.notifyAll();
  94.     }
  96. }
  98. //输出
  100. /*
  101. A=>1
  102. B=>0
  103. A=>1
  104. B=>0
  105. A=>1
  106. B=>0
  107. A=>1
  108. B=>0
  109. A=>1
  110. B=>0
  111. A=>1
  112. B=>0
  113. A=>1
  114. B=>0
  115. A=>1
  116. B=>0
  117. A=>1
  118. B=>0
  119. A=>1
  120. B=>0
  121. C=>1
  122. D=>0
  123. C=>1
  124. D=>0
  125. C=>1
  126. D=>0
  127. C=>1
  128. D=>0
  129. C=>1
  130. D=>0
  131. C=>1
  132. D=>0
  133. C=>1
  134. D=>0
  135. C=>1
  136. D=>0
  137. C=>1
  138. D=>0
  139. C=>1
  140. D=>0
  142. 进程已结束，退出代码为 0
  143. */

Runnable + Lock锁 + condition线程通信方式

  • 使用Lock锁需要先 Lock lock = new ReentrantLock();
  • Lock的线程通信需要先 Condition condition = lock.newCondition(); ,try 语句块加锁 lock.lock(), finally语句块解锁 lock.unlock; ```java import sun.jvm.hotspot.memory.ContiguousSpace;

import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock;

/**

  • Lock方法版

  • 使用OOP类封装属性和同步方法，再去调用Thread线程，传入实例对象的方法开启多线程 */ public class TestJucLock { public static void main(String[] args) {

    1.  // 实例一个需要多线程操作的对象
    2.  Funs fun = new Funs();
    4.  // 将对象需要多线程操作的方法 传给Thread接口开启多线程操作
    5.  new Thread(()->{
    6.      for (int i = 0; i < 10; i++) {
    7.          try {
    8.              fun.increament();
    9.          } catch (InterruptedException e) {
    10.              e.printStackTrace();
    11.          }
    12.      }
    13.  }, "A").start();
    14.  new Thread(()->{
    15.      for (int i = 0; i < 10; i++) {
    16.          try {
    17.              fun.decreament();
    18.          } catch (InterruptedException e) {
    19.              e.printStackTrace();
    20.          }
    21.      }
    22.  }, "B").start();
    24.  new Thread(()->{
    25.      for (int i = 0; i < 10; i++) {
    26.          try {
    27.              fun.increament();
    28.          } catch (InterruptedException e) {
    29.              e.printStackTrace();
    30.          }
    31.      }
    32.  }, "C").start();
    33.  new Thread(()->{
    34.      for (int i = 0; i < 10; i++) {
    35.          try {
    36.              fun.decreament();
    37.          } catch (InterruptedException e) {
    38.              e.printStackTrace();
    39.          }
    40.      }
    41.  }, "D").start();

    } }

class Funs { private int num = 0;

  1. // 使用Lock 需要先在类的作用域定义，不要定义到了方法作用域里面了，
  2. // 不然每次线程调用方法，都会新建一个锁，就锁不住了
  3. Lock lock = new ReentrantLock();
  4. Condition condition = lock.newCondition();
  6. public void increament() throws InterruptedException {
  8.     try {
  9.         lock.lock();
  10.         // 使用while 防止多线程唤醒失败
  11.         while (num == 1) {
  12.             condition.await();
  13.         }
  14.         num++;
  15.         System.out.println(Thread.currentThread().getName() +"=>"+ num);
  16.         condition.signalAll();
  17.     } catch (Exception e) {
  18.         e.printStackTrace();
  19.     } finally {
  20.         lock.unlock();
  21.     }
  23. }
  25. public void decreament() throws InterruptedException {
  27.     try {
  28.         lock.lock();
  29.         // 使用while 防止多线程唤醒失败
  30.         while (num == 0) {
  31.             condition.await();
  32.         }
  33.         num--;
  34.         System.out.println(Thread.currentThread().getName() +"=>"+ num);
  35.         condition.signalAll();
  36.     } catch (Exception e) {
  37.         e.printStackTrace();
  38.     } finally {
  39.         lock.unlock();
  40.     }
  41. }

}

// 输出 // 运行结果与上面类似

  1. <a name="AWYUq"></a>
  2. ### Callbale
  3. ```java
  4. import java.util.concurrent.Callable;
  5. import java.util.concurrent.ExecutionException;
  6. import java.util.concurrent.FutureTask;
  8. public class TestJucCallable {
  9.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  11.         // new Thread(new Runnable()).start()
  12.         // FutureTask是Runnable的实现类
  13.         // new Thread(new FutureTask<V>(Callable)).start()
  14.         MythreadCall call = new MythreadCall();
  15.         FutureTask futureTask = new FutureTask(call);
  17.         new Thread(futureTask, "A").start();
  18.         new Thread(futureTask, "B").start(); // 不会打印call，A的结果会被缓存
  20.         Integer o = (Integer)futureTask.get(); // 返回值
  21.         System.out.println(o);
  22.     }
  23. }
  25. class MythreadCall implements Callable<Integer> {
  27.     @Override
  28.     public Integer call() throws Exception {
  29.         System.out.println("call");
  30.         return 123;
  31.     }
  32. }
  34. // 输出
  35. /*
  36. call
  37. 123
  39. 进程已结束，退出代码为 0
  40. */

线程池

  • 线程池接口：ExecutorService 。 ExecutorService接口继承自Executor接口
  • 线程池的工厂类：Executors（不推荐使用）

  • ThreadPoolExecutor（推荐）

    1.    ExecutorService service = new ThreadPoolExecutor();

    Executors返回的线程池对象的弊端：

  1. FixedThreadPool 和 SingleThreadPolol：

允许的请求队列长度为 Integer.MAX_VALUE（21亿），可能会堆积大量的请求，从而导致OOM。

  1. CachedThreadPool 和 ScheduleThreadPool：

允许的创建线程数量为Integer.MAX_VALUE，可能会堆积大量的线程，从而导致OOM。

  1. import java.util.concurrent.ExecutorService;
  2. import java.util.concurrent.Executors;
  4. public class TestThreadPool{
  5.     public static void main(String[] args) {
  6.         // 创建线程池
  7.         ExecutorService service = Executors.newFixedThreadPool(5);
  8.         for (int i = 1; i <= 10; i++) {
  9.             service.execute(()->{
  10.                 System.out.println(Thread.currentThread().getName());
  11.             });
  12.         }
  13.         // 线程池关闭
  14.         service.shutdown();
  15.     }
  16. }
  18. // 输出，可以看出最多只有5个线程
  19. /*
  20. pool-1-thread-1
  21. pool-1-thread-2
  22. pool-1-thread-3
  23. pool-1-thread-4
  24. pool-1-thread-5 
  25. pool-1-thread-1
  26. pool-1-thread-2
  27. pool-1-thread-3
  28. pool-1-thread-4
  29. pool-1-thread-3
  31. 进程已结束，退出代码为 0
  32. */

JUC并发

集合

  • 普通集合都是线程不安全的
  • List、Map、Set集合都可以使用Collections.synchronizedXXX()进行加锁实现线程安全
  • 直接使用JUC下的线程安全集合 CopyOnWriteArrayList()、CopyOnWriteArraySet()、ConcurrentHashMap()、ConcurrentLinkedDeque()、ConcurrentLinkedQueue()

ConcurrentHashMap()为例：

  1. import java.util.Map;
  2. import java.util.UUID;
  3. import java.util.concurrent.ConcurrentHashMap;
  5. public class TestConcurrentHashMap {
  6.     public static void main(String[] args) {
  7.         // HashMap默认初始值： new HashMap<>(16, 0.75) 16的容量，0.75的装填因子
  8.         Map<String, String> map = new ConcurrentHashMap<>();
  10.         for (int i = 0; i < 30; i++) {
  11.             new Thread(()->{
  12.                 map.put(Thread.currentThread().getName(), UUID.randomUUID().toString().substring(0,5));
  13.                 System.out.println(map);
  14.             }, String.valueOf(i)).start();
  15.         }
  16.     }
  17. }

常用辅助类

CountDownLatch（线程减法计数器）

允许一个或多个线程等待,直到在其他线程中执行的一组操作完成同步辅助(说白了就是一个线程减法计数器)

  1. import java.util.concurrent.CountDownLatch;
  3. public class CountDownLatchDemo {
  4.     // 这里Main函数不应该抛出异常的，但是这里目的是为了简化演示代码就不写try catch了
  5.     public static void main(String[] args) Throws InterruptedException{
  6.         // 线程中的辅助计数器
  7.         CountDownLatch countDownLatch = new CountDownLatch(6);
  9.         for (int i = 1; i <= 6; i++) {
  10.             new Thread(()->{
  11.                 System.out.println(Thread.currentThread().getName()+" Go out");
  12.                 countDownLatch.countDown(); // 计数器-1
  13.             }, String.valueOf(i)).start();
  14.         }
  16.         countDownLatch.await(); // 等待线程计数器归零，才会继续往下执行代码
  17.         System.out.println("Close Door");
  18.     }
  19. }
  21. // 输出
  23. /*
  24. 1 Go out
  25. 2 Go out
  26. 3 Go out
  27. 4 Go out
  28. 5 Go out
  29. 6 Go out
  30. Close Door
  32. 进程已结束，退出代码为 0
  33. */

CyclicBarrier(线程加法计数器)

  1. import java.util.concurrent.BrokenBarrierException;
  2. import java.util.concurrent.CyclicBarrier;
  4. public class CyclicBarrierDemo {
  5.     public static void main(String[] args) {
  6.         /**
  7.          * 用多线程实现 集齐7个龙珠召唤神龙
  8.          */
  10.         // 第一个参数是计数器激活值，第二参数是到达激活值运行的一个Runnable
  11.         CyclicBarrier cyclicBarrier = new CyclicBarrier(7, () -> {
  12.             System.out.println("召唤神龙");
  13.         });
  15.         for (int i = 1; i <= 7; i++) {
  17.             // lambda由于是匿名内部类的进化版，是拿不到外部的for循环的i值的
  18.             // 需要将i值进行finnal处理为一个常量
  19.             final int temp = i;
  21.             new Thread(()->{
  22.                 System.out.println(Thread.currentThread().getName()+ "==>收集第" + temp + "个龙珠");
  23.                 try {
  24.                     // 线程等待,直到满足线程计数设定值才会往下继续执行
  25.                     // 返回值是当前线程计数值，每个新线程会自动加一
  26.                     int await = cyclicBarrier.await();
  28.                     // 召唤神龙之后才会打印,多线程打印，随机顺序
  29.                     System.out.println(await); 
  31.                 } catch (InterruptedException e) {
  32.                     e.printStackTrace();
  33.                 } catch (BrokenBarrierException e) {
  34.                     e.printStackTrace();
  35.                 }
  36.             }).start();
  37.         }
  38.     }
  39. }
  41. // 输出
  43. /*
  44. Thread-0==>收集第1个龙珠
  45. Thread-1==>收集第2个龙珠
  46. Thread-2==>收集第3个龙珠
  47. Thread-3==>收集第4个龙珠
  48. Thread-4==>收集第5个龙珠
  49. Thread-5==>收集第6个龙珠
  50. Thread-6==>收集第7个龙珠
  51. 召唤神龙
  52. 0
  53. 4
  54. 3
  55. 2
  56. 5
  57. 6
  58. 1
  60. 进程已结束，退出代码为 0
  61. */

CyclicBarrier和CountDownLatch的区别

  • CountDownLatch是减法计数器，CyclicBarrier是加法计数器。
  • CountDownLatch的计数器只能使用一次，而CyclicBarrier的计数器可以使用reset()方法重置。CountDownLatch和CyclicBarrier都有让多个线程等待同步然后再开始下一步动作的意思，但是CountDownLatch的下一步的动作实施者是主线程，具有不可重复性；而CyclicBarrier的下一步动作实施者还是“其他线程”本身，具有往复多次实施动作的特点。

  • CyclicBarrier还提供其他有用的方法，getNumberWaiting()方法可以获得阻塞线程的数量；isBroken()方法可以了解阻塞的线程是否被中断。

    Semaphore（规定线程数量，限流作用）

    有点wait()、notifyAll()的味道，但是是自动计算阈值的，并且控制的是线程

  • acquire() 获取，也就是激活一个线程，直到满足设定的线程最大数量时进入等待状态

  • release() 释放，释放当前的线程，线程数量+1，然后唤醒等待线程

作用

  1. 多个共享资源互斥的使用
  2. 并发限流，控制最大线程数 ```java import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit;

public class SemaphoreDemo { /* 

  1.  * 完成停车位功能
  2.  * @param args
  3.  */
  4. public static void main(String[] args) {
  5.     // 参数就是信号量，也就是限制线程数量
  6.     Semaphore semaphore = new Semaphore(3);
  8.     for (int i = 1; i <=6 ; i++) {
  9.         // 只会运行3个线程运行
  10.         new Thread(()->{
  11.             try {
  13.                 // 获取，也就是激活一个线程
  14.                 semaphore.acquire();
  15.                 System.out.println(Thread.currentThread().getName()+"==>进入车位");
  17.                 // 线程等待2s
  18.                 TimeUnit.SECONDS.sleep(1);
  20.                 System.out.println(Thread.currentThread().getName()+"==>离开车位");
  21.             } catch (InterruptedException e) {
  22.                 e.printStackTrace();
  23.             }
  24.             finally {
  25.                 semaphore.release(); // 释放线程
  26.             }
  27.         }).start();
  28.     }
  29. }

}

/* 输出：

Thread-0==>进入车位 Thread-2==>进入车位 Thread-1==>进入车位 Thread-1==>离开车位 Thread-2==>离开车位 Thread-3==>进入车位 Thread-0==>离开车位 Thread-4==>进入车位 Thread-5==>进入车位 Thread-5==>离开车位 Thread-3==>离开车位 Thread-4==>离开车位

进程已结束，退出代码为 0

*/

  1. <a name="AARPh"></a>
  2. ### 读写锁
  3. <a name="Pv4zV"></a>
  4. #### ReadWriteLock
  5. 作用：
  7. - 比ReentrantLock有更小的细粒度，可以对读写分别加锁，并且写入锁是**排他锁**，读取锁是**共享锁**
  8. - 在需要加锁的类的属性域中定义：**ReadWriteLock readWriteLock = new ReentrantReadWriteLock();**
  10. 在加锁方法中**try块**进行加锁**readWriteLock.writeLock().lock();**<br />**finally块**进行解锁 **readWriteLock.writeLock().unlock()；**
  11. ```java
  12. import java.util.HashMap;
  13. import java.util.Map;
  14. import java.util.concurrent.locks.ReadWriteLock;
  15. import java.util.concurrent.locks.ReentrantReadWriteLock;
  17. public class ReadWriteLockDemo {
  18.     public static void main(String[] args) {
  19.         MyCacheLock cache = new MyCacheLock();
  21.         // 写入
  22.         for (int i = 1; i <= 5; i++) {
  23.             final int temp = i;
  24.             new Thread(()->{
  25.                 cache.put(temp+"",temp);
  26.             }).start();
  27.         }
  29.         System.out.println("========="); // main主线程，只会执行一次，而且与其他线程抢夺资源
  30.         // 读取
  31.         for (int i = 1; i <= 5; i++) {
  32.             final int temp = i;
  33.             new Thread(()->{
  34.                 cache.get(temp+"");
  35.             }).start();
  36.         }
  37.         System.out.println("=========");
  38.     }
  39. }
  41. // 设计一个缓存
  42. /*
  43. 这里提一句LRU缓存，使用HashMap + 双向链表实现（其实可以直接继承LinkedHashMap），
  44. get和set都将缓存对象提升到链表头，当缓存满了的时候，就删掉链表尾部的缓存
  45.  */
  46. class MyCacheLock{
  48.     private volatile Map<String, Integer> map = new HashMap<>();
  49.     private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
  51.     // 存入
  52.     public void put(String key, int value) {
  53.         try {
  54.             readWriteLock.writeLock().lock();
  56.             System.out.println(Thread.currentThread().getName()+ "==>写入" + key);
  57.             map.put(key, value);
  58.             System.out.println(Thread.currentThread().getName()+ "==>写入完毕，ok");
  59.         } catch (Exception e) {
  60.             e.printStackTrace();
  61.         } finally {
  62.             readWriteLock.writeLock().unlock();
  63.         }
  64.     }
  66.     // 读取
  67.     public void get(String key) {
  68.         try {
  69.             readWriteLock.readLock().lock();
  71.             System.out.println(Thread.currentThread().getName()+ "==>读取" + key);
  72.             Integer i = map.get(key);
  73.             System.out.println(Thread.currentThread().getName()+ "==>读取完毕，ok");
  74.         } catch (Exception e) {
  75.             e.printStackTrace();
  76.         } finally {
  77.             readWriteLock.readLock().unlock();
  78.         }
  80.     }
  81. }
  83. // 输出
  85. /*
  86. Thread-0==>写入1
  87. Thread-0==>写入完毕，ok
  88. Thread-1==>写入2
  89. Thread-1==>写入完毕，ok
  90. Thread-2==>写入3
  91. Thread-2==>写入完毕，ok
  92. Thread-3==>写入4
  93. Thread-3==>写入完毕，ok
  94. Thread-4==>写入5
  95. Thread-4==>写入完毕，ok
  96. =========
  97. Thread-5==>读取1
  98. Thread-5==>读取完毕，ok
  99. Thread-6==>读取2
  100. Thread-6==>读取完毕，ok
  101. Thread-7==>读取3
  102. Thread-7==>读取完毕，ok
  103. Thread-8==>读取4
  104. Thread-8==>读取完毕，ok
  105. =========
  106. Thread-9==>读取5
  107. Thread-9==>读取完毕，ok
  109. 进程已结束，退出代码为 0
  111. */

阻塞队列

BlockingQueue（实现类：ArrayBlockingQueue、LinkedBlockingQueue、同步队列SynchronousQueue）

image.png
四组API：
image.png

SynchronousQueue

同步队列，put了一个元素就必须等待take取出来，不然不能再put任何元素！

  1. import java.util.concurrent.BlockingDeque;
  2. import java.util.concurrent.BlockingQueue;
  3. import java.util.concurrent.SynchronousQueue;
  4. import java.util.concurrent.TimeUnit;
  6. public class SynchronousQueueDemo {
  7.     public static void main(String[] args) {
  8.         BlockingQueue<String> synQueue = new SynchronousQueue<>();
  10.         new Thread(()->{
  12.             try {
  13.                 System.out.println(Thread.currentThread().getName() + " put 1");
  14.                 synQueue.put("1");
  15.                 System.out.println(Thread.currentThread().getName() + " put 2");
  16.                 synQueue.put("2");
  17.                 System.out.println(Thread.currentThread().getName() + " put 3");
  18.                 synQueue.put("3");
  19.             } catch (InterruptedException e) {
  20.                 e.printStackTrace();
  21.             }
  22.         }, "T1").start();
  24.         new Thread(()->{
  26.             try {
  27.                 TimeUnit.SECONDS.sleep(1);
  28.                 System.out.println(Thread.currentThread().getName() + " take " + synQueue.take());
  29.                 TimeUnit.SECONDS.sleep(1);
  30.                 System.out.println(Thread.currentThread().getName() + " take " + synQueue.take());
  31.                 TimeUnit.SECONDS.sleep(1);
  32.                 System.out.println(Thread.currentThread().getName() + " take " + synQueue.take());
  33.             } catch (InterruptedException e) {
  34.                 e.printStackTrace();
  35.             }
  36.         }, "T2").start();
  37.     }
  38. }
  39. // 输出
  40. /*
  41. T1 put 1
  42. T2 take 1
  43. T1 put 2
  44. T2 take 2
  45. T1 put 3
  46. T2 take 3
  48. 进程已结束，退出代码为 0
  49. */

线程池（池化技术）

ExecutorService service = new ThreadPoolExecutor();

线程的 3大方法，7大参数。

3大方法：（都是ThreadPoolExecutor()的不同参数，不推荐使用；推荐直接使用ThreadPoolExecutor()）

Executors.newFixedThreadPool(int nThreads); // 创建一个固定的线程池大小
Executors.newCachedThreadPool(); // 可伸缩的线程池
Executors.newSingleThreadExecutor(); // 单个线程
源码：

  1. // newFixedThreadPool
  2. public static ExecutorService newFixedThreadPool(int nThreads) {
  3.     return new ThreadPoolExecutor(nThreads, nThreads,
  4.                                   0L, TimeUnit.MILLISECONDS,
  5.                                   new LinkedBlockingQueue<Runnable>());
  6. }
  8. // newCachedThreadPool
  9. public static ExecutorService newCachedThreadPool() {
  10.     return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
  11.                                   60L, TimeUnit.SECONDS,
  12.                                   new SynchronousQueue<Runnable>());
  13. }
  15. // newSingleThreadExecutor
  16. public static ExecutorService newSingleThreadExecutor() {
  17.     return new FinalizableDelegatedExecutorService
  18.         (new ThreadPoolExecutor(1, 1,
  19.                                 0L, TimeUnit.MILLISECONDS,
  20.                                 new LinkedBlockingQueue<Runnable>()));
  21. }

通过源码可以看出，3大方法的本质就是调用了 ThreadPoolExecutor(),

7大参数 就是指的ThreadPoolExecutor的参数
  1. public ThreadPoolExecutor(int corePoolSize, // 核心线程池大小
  2.                               int maximumPoolSize, // 最大核心线程池大小
  3.                               long keepAliveTime, // 无调用，超时释放
  4.                               TimeUnit unit, // 超时单位
  5.                               BlockingQueue<Runnable> workQueue, // 阻塞队列
  6.                               ThreadFactory threadFactory, // 线程工厂
  7.                               RejectedExecutionHandler handler) // 拒绝策略

image.png
四种拒绝策略 最大线程定义准则（CPU密集型、IO密集型）

  1. import java.util.concurrent.*;
  3. /**
  4.  * 实现银行业务办理
  5.  */
  6. public class ThreadPoolExecutorDemo {
  7.     public static void main(String[] args) {
  8.         // 最大线程定义准则：
  9.         // 1、CPU密集型
  10.         // 2、IO密集型： 定义为程序中非常消耗IO的线程 的两倍
  12.         // 获取CPU核数
  13.         int i1 = Runtime.getRuntime().availableProcessors();
  14.         ExecutorService threadPool = new ThreadPoolExecutor(
  15.                 2, // 核心业务窗口数量
  16.                 i1, // 最大核心业务窗口数量
  17.                 3, // 超时释放时间
  18.                 TimeUnit.SECONDS, // 超时单位
  19.                 new LinkedBlockingDeque<>(3), // 阻塞队列大小
  20.                 Executors.defaultThreadFactory(), // 默认工厂类
  21.                 new ThreadPoolExecutor.AbortPolicy() // 队列满了还有元素进来，不处理并且抛出异常
  22.                 //new ThreadPoolExecutor.CallerRunsPolicy() // 回自到己的线程去
  23.                 //new ThreadPoolExecutor.DiscardPolicy() // 队列满了，丢掉任务，不抛出异常
  24.                 //new ThreadPoolExecutor.DiscardOldestPolicy() // 队列满了，尝试和最早的竞争，不会抛出异常
  25.         );
  27.         try {
  28.             for (int i = 1; i <= 9; i++) {
  29.                 threadPool.execute(() -> {
  30.                     System.out.println(Thread.currentThread().getName());
  31.                 });
  32.             }
  33.         } catch (Exception e) {
  34.             e.printStackTrace();
  35.         } finally {
  37.             threadPool.shutdown();
  38.         }
  40.     }
  41. }

四大函数式接口

Function、Predict、Consumer、Supplier

  • Function 函数型接口：范型第一个参数是输入类型，第二个参数是输出类型

image.png

  1. Function<String, String> function = (str)->{return str;};
  3. System.out.println(function.apply("aaa")) // out: "aaa"
  • Predict 断定型接口：一个输入参数，返回只能是 布尔值

image.png

  1. Predict<String> predict= (str)->{return str.isEmpty();};
  3. System.out.println(predict.test("")) // out: true
  • Consumer 消费型接口：只有输入，没有返回值

image.png

  1. Consumer<String> consumer = (str)->{System.out.println(str);};
  3. consumer.accept("asdasda"); // out: asdasda
  • Supplier 供给型接口：没有参数，只有返回值

image.png

  1. Supplier<> supplier = ()->{return 1024;};
  3. System.out.println(supplier.get()); // out: 1024

Stream流式计算

image.png

执行机制：

image.png

  1. users.stream()
  2.     .filter(u -> u.getId()%2==0) // 过滤器
  3.     .map(u -> u.getAge()+1) // 映射器
  4.     //.count(); // 计数器
  5.     //.sorted() // 排序
  6.     //.min()    // 求最小值
  7.     //.limit()   // 分页
  8.     //.collect(Collectors.toList()) // 将流转换成集合和聚合元素。Collectors 可用于返回列表或字符串
  9.     .forEach(System.out::println);

ForkJoin

  • 将大任务拆成小任务

image.png

  • 计算类继承ForkJoinTask类，一般继承下面的两个子类（RecursiveAction、RecursiveTask）

image.png

  • 计算类代码 ```java package stream;

import java.util.concurrent.RecursiveTask;

public class ForkJoinDemo extends RecursiveTask {

  1. private Long start;
  2. private Long end;
  4. // 临界值
  5. private Long temp = 10000L;
  7. public ForkJoinDemo(Long start, Long end) {
  8.     this.start = start;
  9.     this.end = end;
  10. }
  13. @Override
  14. protected Long compute() {
  16.     if ((end - start) < temp) {
  17.         Long sum = 0L;
  18.         for (Long i = start; i <= end; ++i) {
  19.             sum += i;
  20.         }
  21.         return sum;
  22.     }
  25.     // ForkJoin 递归计算
  26.     // 任务拆分
  27.     Long mid = (start + end) / 2; // 中间值
  29.     // 任务1
  30.     ForkJoinDemo task1 = new ForkJoinDemo(start, mid);
  31.     task1.fork(); // 把任务压入线程队列中
  33.     // 任务2
  34.     ForkJoinDemo task2 = new ForkJoinDemo(mid+1, end);
  35.     task2.fork(); // 把任务压入线程队列中
  37.     // 结果合并
  38.     return task1.join() + task2.join();
  40. }

}

  2. - 测试类代码
  3. ```java
  4. package stream;
  6. import java.util.concurrent.ExecutionException;
  7. import java.util.concurrent.ForkJoinPool;
  8. import java.util.concurrent.ForkJoinTask;
  9. import java.util.stream.LongStream;
  11. public class TestForkJoin {
  12.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  13.         //test1(); // 执行时间8122
  14.         test2(); // 执行时间7944
  15.         //test3(); // 执行时间499
  16.     }
  18.     public static void test1() {
  20.         Long sum =0L;
  22.         long start = System.currentTimeMillis();
  24.         for (Long i = 1L; i <= 10_0000_0000L; ++i)
  25.             sum += i;
  27.         long end = System.currentTimeMillis();
  28.         System.out.println("sum="+ sum +"  执行时间"+(end - start)); // 8276
  29.     }
  31.     // ForkJoin
  32.     public static void test2() throws ExecutionException, InterruptedException {
  33.         long start = System.currentTimeMillis();
  35.         ForkJoinPool forkJoinPool = new ForkJoinPool();
  36.         ForkJoinTask<Long> task = new ForkJoinDemo(1L, 10_0000_0000L);
  37.         ForkJoinTask<Long> submit = forkJoinPool.submit(task);
  39.         Long sum = submit.get();
  41.         long end = System.currentTimeMillis();
  42.         System.out.println("sum="+ sum +"  执行时间"+(end - start)); // 130
  43.     }
  45.     // Stram并行流
  46.     public static void test3() {
  47.         long start = System.currentTimeMillis();
  49.         long sum = LongStream.rangeClosed(0L, 10_0000_0000L).parallel().reduce(0, Long::sum);
  50.         long end = System.currentTimeMillis();
  51.         System.out.println("sum="+ sum +"  执行时间"+(end - start)); // 483
  52.     }
  53. }

异步回调

使用Future的实现类CompletableFuture

  1. import java.util.concurrent.CompletableFuture;
  2. import java.util.concurrent.ExecutionException;
  3. import java.util.concurrent.TimeUnit;
  5. /**
  6.  * 异步调用： CompletableFuture
  7.  * 异步执行
  8.  * 成功回调
  9.  * 失败回调
  10.  */
  11. public class AsynDemo {
  13.     public static void main(String[] args) throws Exception {
  14. //        // 异步执行，没有返回值
  15. //        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(()->{
  16. //            try {
  17. //                TimeUnit.SECONDS.sleep(1);
  18. //            } catch (InterruptedException e) {
  19. //                e.printStackTrace();
  20. //            }
  21. //
  22. //            System.out.println(Thread.currentThread().getName()+"runAsync => Void");
  23. //        });
  24. //
  25. //        // 不管异步线程，继续执行Main线程
  26. //        System.out.println("111");
  27. //
  28. //        voidCompletableFuture.get(); // 获取异步执行结果
  31.         // 有返回值调
  32.         CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
  33.             try {
  34.                 TimeUnit.SECONDS.sleep(1);
  35.                 System.out.println(Thread.currentThread().getName()+ "supplyAsync=>Integer");
  36.                 //int i = 1/0;
  37.             } catch (InterruptedException e) {
  38.                 e.printStackTrace();
  39.             }
  40.             return 1024;
  41.         });
  43.         System.out.println("111");
  46.         System.out.println(completableFuture.whenComplete((t, u) -> {
  47.             System.out.println("t=> " + t); // 第一个参数：成功 异步回调调返回值， 失败 null
  48.             System.out.println("u=> " + u); // 第二个参数：成功nul，失败 错误返回值
  49.         }).exceptionally((e) -> {
  50.             System.out.println(e.getMessage());
  51.             return 233;
  52.         }).get()); // 成功：1024，失败：233
  53.     }
  54. }

JMM

彻底明白单例模式