• Java多线程及原理

    工作原理

    线程池内部是通过队列结合线程实现的，当我们利用线程池执行任务时：

    1. 如果此时线程池中的线程数量小于corePoolSize，即使线程池中的线程都处于空闲状态，也要创建新的线程来处理被添加的任务。

    2. 如果此时线程池中的线程数量等于corePoolSize，但是缓冲队列workQueue未满，那么任务被放入缓冲队列。

    3. 如果此时线程池中的线程数量大于等于corePoolSize，缓冲队列workQueue已满，并且线程池中的线程数量小于maximumPoolSize，建新的线程来处理被添加的任务。

    4. 如果此时线裎池中的线数量大于corePoolSize，缓存冲队列workQueue已满, 并且线程池中的数量等于maximumPoolSize，那么过handler所指定的策略来处理此任务。

    5. 当线程池中的线程数量大于corePoolSize时，如果某线程空闲时间超过keepAliveTime, 线将被终止。这样，线程池可以动态的调整池中的线程数。

    相关配置

    corePoolSize：核心线程数

    maximumPoolSize：最大线程数 【包括核心线程数】

    keepAliveTime：生存时间【线程长时间不干活了，归还给操作系统，核心线程不用归还，可以指定是否参与归还过程】

    生存时间单位

    任务队列：等待队列，如果不指定，最大值是Integer.MAX_VALUE【各种各样的BlockingQueue】

    线程工厂【默认设置优先级是普通优先级，非守护线程】，最好自定义线程名称，方便回溯

    拒绝策略，包括以下四种：

    ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。

    ThreadPoolExecutor.DiscardPolicy：丢弃任务，但是不抛出异常。

    ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新提交被拒绝的任务

    ThreadPoolExecutor.CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务

    执行流程：先占满核心线程-> 再占满任务队列-> 再占满（最大线程数-核心线程数）-> 最后执行拒绝策略
    一般自定义拒绝策略：将相关信息保存到redis，kafka，日志，MySQL记录 实现RejectedExecutionHandler并重写rejectedExecution方法

    自定义拒绝策略代码示例

    1. package git.snippets.juc;
    2.  
    3. import java.util.concurrent.*;
    4.  
    5. /**
    6.  * 自定义拒绝策略
    7.  */
    8. public class MyRejectedHandler {
    9.     public static void main(String[] args) {
    10.         ExecutorService service = new ThreadPoolExecutor(4, 4,
    11.                 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(6),
    12.                 Executors.defaultThreadFactory(),
    13.                 new MyHandler());
    14.     }
    15.  
    16.     static class MyHandler implements RejectedExecutionHandler {
    17.  
    18.         @Override
    19.         public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
    20.             //log("r rejected")
    21.             //save r kafka mysql redis
    22.             //try 3 times
    23.             if (executor.getQueue().size() < 10000) {
    24.                 //try put again();
    25.             }
    26.         }
    27.     }
    28. }
    29.

    SingleThreadPool

    • 保证线程按顺序执行

    • 为什么要有单线程的线程池？这个主要是用来做任务队列和线程生命周期管理

    • 使用LinkedBlockingQueue作为任务队列，上界为：Integer.MAX_VALUE(2147483647) 约等于无界。

    示例代码：

    1. package git.snippets.juc;
    2.  
    3. import java.util.concurrent.ExecutorService;
    4. import java.util.concurrent.Executors;
    5.  
    6. import static java.util.concurrent.TimeUnit.SECONDS;
    7.  
    8. public class SingleThreadPoolUsage {
    9.     public static void main(String[] args) throws InterruptedException {
    10.         ExecutorService service = Executors.newSingleThreadExecutor();
    11.         for (int i = 0; i < 10; i++) {
    12.             final int j = i;
    13.             service.submit(() -> System.out.println("current thread " + Thread.currentThread() + "  " + j));
    14.         }
    15.         service.shutdown();
    16.         service.awaitTermination(60, SECONDS);
    17.     }
    18. }
    19.

    CachedThreadPool

    • corePoolSize：0

    • maxiumPoolSize：Integer.MAX_VALUE(2147483647)

    • keepAliveTime 60秒

    • 使用SynchronousQueue作为任务队列 必须马上执行

    使用示例：

    1. package git.snippets.juc;
    2.  
    3. import java.util.concurrent.ExecutorService;
    4. import java.util.concurrent.Executors;
    5. import java.util.concurrent.TimeUnit;
    6.  
    7. public class CachedThreadPoolUsage {
    8.     public static void main(String[] args) throws InterruptedException {
    9.         System.out.println("cached thread pool usage...");
    10.         ExecutorService service = Executors.newCachedThreadPool();
    11.         System.out.println(service);
    12.         for (int i = 0; i < 2; i++) {
    13.             service.execute(() -> {
    14.                 try {
    15.                     TimeUnit.MILLISECONDS.sleep(500);
    16.                 } catch (InterruptedException e) {
    17.                     e.printStackTrace();
    18.                 }
    19.                 System.out.println(Thread.currentThread().getName());
    20.             });
    21.         }
    22.         System.out.println(service);
    23.         TimeUnit.SECONDS.sleep(80);
    24.         System.out.println(service);
    25.     }
    26. }
    27.

    FixedThreadPool

    • 最大线程数等于核心线程数

    • 使用LinkedBlockingQueue作为任务队列，上界为：Integer.MAX_VALUE(2147483647)

    使用示例见：

    1. package git.snippets.juc;
    2.  
    3. import java.util.ArrayList;
    4. import java.util.List;
    5. import java.util.concurrent.Callable;
    6. import java.util.concurrent.ExecutionException;
    7. import java.util.concurrent.ExecutorService;
    8. import java.util.concurrent.Executors;
    9. import java.util.concurrent.Future;
    10. import java.util.concurrent.TimeUnit;
    11.  
    12. /**
    13.  * 多线程和单线程计算某个范围内的所有素数
    14.  */
    15. public class FixedThreadPoolUsage {
    16.     public static void main(String[] args) throws InterruptedException, ExecutionException {
    17.         long start = System.currentTimeMillis();
    18.         getPrime(1, 200000);
    19.         long end = System.currentTimeMillis();
    20.         System.out.println("use single thread...cost: " + (end - start));
    21.  
    22.         final int cpuCoreNum = 4;
    23.  
    24.         ExecutorService service = Executors.newFixedThreadPool(cpuCoreNum);
    25.  
    26.         MyTask t1 = new MyTask(1, 80000); //1-5 5-10 10-15 15-20
    27.         MyTask t2 = new MyTask(80001, 130000);
    28.         MyTask t3 = new MyTask(130001, 170000);
    29.         MyTask t4 = new MyTask(170001, 200000);
    30.  
    31.         Future> f1 = service.submit(t1);
    32.         Future> f2 = service.submit(t2);
    33.         Future> f3 = service.submit(t3);
    34.         Future> f4 = service.submit(t4);
    35.         System.out.println();
    36.         start = System.currentTimeMillis();
    37.         f1.get();
    38.         f2.get();
    39.         f3.get();
    40.         f4.get();
    41.         end = System.currentTimeMillis();
    42.         System.out.println("use fixed thread pool...cost: " + (end - start));
    43.         service.shutdown();
    44.         service.awaitTermination(1, TimeUnit.MINUTES);
    45.     }
    46.  
    47.     static boolean isPrime(int num) {
    48.         for (int i = 2; i <= num / 2; i++) {
    49.             if (num % i == 0) {
    50.                 return false;
    51.             }
    52.         }
    53.         return true;
    54.     }
    55.  
    56.     static List getPrime(int start, int end) {
    57.         List results = new ArrayList<>();
    58.         for (int i = start; i <= end; i++) {
    59.             if (isPrime(i)) results.add(i);
    60.         }
    61.  
    62.         return results;
    63.     }
    64.  
    65.     static class MyTask implements Callable> {
    66.         int startPos, endPos;
    67.  
    68.         MyTask(int s, int e) {
    69.             this.startPos = s;
    70.             this.endPos = e;
    71.         }
    72.  
    73.         @Override
    74.         public List call() {
    75.             List r = getPrime(startPos, endPos);
    76.             return r;
    77.         }
    78.  
    79.     }
    80. }

    ScheduledThreadPool

    使用DelayWorkQueue，包括了如下两个主要方法

    scheduleAtFixedRate()

    当前任务执行时间小于间隔时间，每次到点即执行；

    当前任务执行时间大于等于间隔时间，任务执行后立即执行下一次任务。相当于连续执行了。

    scheduleWithFixedDelay()

    每当上次任务执行完毕后，间隔一段时间执行。不管当前任务执行时间大于、等于还是小于间隔时间，执行效果都是一样的。

    使用示例：

    1. package git.snippets.juc;
    2.  
    3. import java.util.Date;
    4. import java.util.concurrent.Executors;
    5. import java.util.concurrent.ScheduledExecutorService;
    6.  
    7. import static java.util.concurrent.TimeUnit.SECONDS;
    8.  
    9. public class ScheduleThreadPoolUsage {
    10.     static ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    11.  
    12.  
    13.     public static void main(String[] args) {
    14.         test1();
    15.         test2();
    16.         test3();
    17.     }
    18.  
    19.     /**
    20.      * 任务执行时间（8s）小于间隔时间(10s)
    21.      */
    22.     public static void test1() {
    23.         scheduler.scheduleAtFixedRate(() -> {
    24.             System.out.println("Start: scheduleAtFixedRate:    " + new Date());
    25.             try {
    26.                 Thread.sleep(8000);
    27.             } catch (InterruptedException e) {
    28.                 e.printStackTrace();
    29.             }
    30.             System.out.println("End  : scheduleAtFixedRate:    " + new Date());
    31.         }, 0, 10, SECONDS);
    32.     }
    33.  
    34.     /**
    35.      * 任务执行时间（12s）大于间隔时间(10s)
    36.      */
    37.     public static void test2() {
    38.         scheduler.scheduleAtFixedRate(() -> {
    39.             System.out.println("Start: scheduleAtFixedRate:    " + new Date());
    40.             try {
    41.                 Thread.sleep(12000);
    42.             } catch (InterruptedException e) {
    43.                 e.printStackTrace();
    44.             }
    45.             System.out.println("End  : scheduleAtFixedRate:    " + new Date());
    46.         }, 0, 10, SECONDS);
    47.     }
    48.  
    49.     /**
    50.      * 任务执行时间（8s）小于间隔时间(10s)
    51.      */
    52.     public static void test3() {
    53.         scheduler.scheduleWithFixedDelay(() -> {
    54.             System.out.println("Start: scheduleWithFixedDelay: " + new Date());
    55.             try {
    56.                 Thread.sleep(12000);
    57.             } catch (InterruptedException e) {
    58.                 e.printStackTrace();
    59.             }
    60.             System.out.println("End  : scheduleWithFixedDelay: " + new Date());
    61.         }, 0, 10, SECONDS);
    62.     }
    63. }
    64.

    ForkJoinPool

    Java SE 1.7 以后新增的线程池，包括以下两个核心类

    第一个是：RecursiveAction

    它是一种没有任何返回值的任务。只是做一些工作，比如写数据到磁盘，然后就退出了。 一个RecursiveAction可以把自己的工作分割成更小的几块， 这样它们可以由独立的线程或者 CPU 执行。
    我们可以通过继承来实现一个RecursiveAction。

    第二个是：RecursiveTask

    它是一种会返回结果的任务。可以将自己的工作分割为若干更小任务，并将这些子任务的执行合并到一个集体结果。 可以有几个水平的分割和合并。

    使用示例：

    1. package git.snippets.juc;
    2.  
    3. import java.util.concurrent.ForkJoinPool;
    4. import java.util.concurrent.RecursiveAction;
    5. import java.util.concurrent.RecursiveTask;
    6. import java.util.concurrent.TimeUnit;
    7. import java.util.stream.LongStream;
    8.  
    9.  
    10. public class ForkJoinPoolUsage implements Calculator {
    11.     private ForkJoinPool pool;
    12.  
    13.     public ForkJoinPoolUsage() {
    14.         // 也可以使用公用的 ForkJoinPool：
    15.         // pool = ForkJoinPool.commonPool()
    16.         pool = new ForkJoinPool();
    17.     }
    18.  
    19.     public static void useRecursiveAction() throws InterruptedException {
    20.         // 创建包含Runtime.getRuntime().availableProcessors()返回值作为个数的并行线程的ForkJoinPool
    21.         ForkJoinPool forkJoinPool = new ForkJoinPool();
    22.         // 提交可分解的PrintTask任务
    23.         forkJoinPool.submit(new MyRecursiveAction(0, 1000));
    24.  
    25.         while (!forkJoinPool.isTerminated()) {
    26.             forkJoinPool.awaitTermination(2, TimeUnit.SECONDS);
    27.         }
    28.         // 关闭线程池
    29.         forkJoinPool.shutdown();
    30.     }
    31.  
    32.     public static void useRecursiveTask() {
    33.         long[] numbers = LongStream.rangeClosed(1, 1000).toArray();
    34.         Calculator calculator = new ForkJoinPoolUsage();
    35.         System.out.println(calculator.sumUp(numbers)); // 打印结果500500
    36.     }
    37.  
    38.     public static void main(String[] args) throws InterruptedException {
    39.         useRecursiveTask();
    40.         useRecursiveAction();
    41.     }
    42.  
    43.     @Override
    44.     public long sumUp(long[] numbers) {
    45.         return pool.invoke(new SumTask(numbers, 0, numbers.length - 1));
    46.     }
    47.  
    48.     private static class MyRecursiveAction extends RecursiveAction {
    49.  
    50.         /**
    51.          * 每个"小任务"最多只打印20个数
    52.          */
    53.         private static final int MAX = 20;
    54.  
    55.         private int start;
    56.         private int end;
    57.  
    58.         public MyRecursiveAction(int start, int end) {
    59.             this.start = start;
    60.             this.end = end;
    61.         }
    62.  
    63.         @Override
    64.         protected void compute() {
    65.             //当end-start的值小于MAX时，开始打印
    66.             if ((end - start) < MAX) {
    67.                 for (int i = start; i < end; i++) {
    68.                     System.out.println(Thread.currentThread().getName() + "-i的值" + i);
    69.                 }
    70.             } else {
    71.                 // 将大任务分解成两个小任务
    72.                 int middle = (start + end) / 2;
    73.                 MyRecursiveAction left = new MyRecursiveAction(start, middle);
    74.                 MyRecursiveAction right = new MyRecursiveAction(middle, end);
    75.                 left.fork();
    76.                 right.fork();
    77.             }
    78.         }
    79.  
    80.  
    81.     }
    82.  
    83.     private static class SumTask extends RecursiveTask {
    84.         private long[] numbers;
    85.         private int from;
    86.         private int to;
    87.  
    88.         public SumTask(long[] numbers, int from, int to) {
    89.             this.numbers = numbers;
    90.             this.from = from;
    91.             this.to = to;
    92.         }
    93.  
    94.  
    95.         @Override
    96.         protected Long compute() {
    97.  
    98.             // 当需要计算的数字小于6时，直接计算结果
    99.             if (to - from < 6) {
    100.                 long total = 0;
    101.                 for (int i = from; i <= to; i++) {
    102.                     total += numbers[i];
    103.                 }
    104.                 return total;
    105.                 // 否则，把任务一分为二，递归计算
    106.             } else {
    107.                 int middle = (from + to) / 2;
    108.                 SumTask taskLeft = new SumTask(numbers, from, middle);
    109.                 SumTask taskRight = new SumTask(numbers, middle + 1, to);
    110.                 taskLeft.fork();
    111.                 taskRight.fork();
    112.                 return taskLeft.join() + taskRight.join();
    113.             }
    114.         }
    115.     }
    116. }
    117.  
    118. interface Calculator {
    119.     long sumUp(long[] numbers);
    120. }

    WorkStealingPool

    每个线程都有单独的队列，每个线程队列执行完毕后，就会去其他的线程队列里面拿过来执行, 底层是ForkJoinPool

    • Java SE 1.8 新增

    • 会自动启动 CPU 核数个线程去执行任务

    使用示例：

    1. /**
    2.  *
    3.  */
    4. package git.snippets.juc;
    5.  
    6. import java.io.IOException;
    7. import java.util.concurrent.ExecutorService;
    8. import java.util.concurrent.Executors;
    9. import java.util.concurrent.TimeUnit;
    10.  
    11. /**
    12.  * @since 1.8
    13.  */
    14. public class WorkStealingPoolUsage {
    15.     public static void main(String[] args) throws IOException {
    16.         int core = Runtime.getRuntime().availableProcessors();
    17.         //  会自动启动cpu核数个线程去执行任务 ,其中第一个是1s执行完毕,其余都是2s执行完毕,
    18.         //  有一个任务会进行等待,当第一个执行完毕后,会再次偷取最后一个任务执行
    19.         ExecutorService service = Executors.newWorkStealingPool();
    20.         service.execute(new R(1000));
    21.         for (int i = 0; i < core; i++) {
    22.             service.execute(new R(2000));
    23.         }
    24.         //由于产生的是精灵线程（守护线程、后台线程），主线程不阻塞的话，看不到输出
    25.         System.in.read();
    26.     }
    27.  
    28.     static class R implements Runnable {
    29.  
    30.         int time;
    31.  
    32.         R(int t) {
    33.             this.time = t;
    34.         }
    35.  
    36.         @Override
    37.         public void run() {
    38.             try {
    39.                 TimeUnit.MILLISECONDS.sleep(time);
    40.             } catch (InterruptedException e) {
    41.                 e.printStackTrace();
    42.             }
    43.  
    44.             System.out.println(time + " " + Thread.currentThread().getName());
    45.         }
    46.     }
    47. }
    48.

    CompletableFuture

    • Java SE 1.8 新增

    • anyOf()可以实现“任意个 CompletableFuture 只要一个成功”，allOf()可以实现“所有 CompletableFuture 都必须成功”，这些组合操作可以实现非常复杂的异步流程控制。

    使用示例：

    1. package git.snippets.juc;
    2.  
    3. import java.util.Random;
    4. import java.util.concurrent.CompletableFuture;
    5. import java.util.concurrent.ExecutionException;
    6. import java.util.concurrent.TimeUnit;
    7.  
    8. /**
    9.  * 假设你能够提供一个服务
    10.  * 这个服务查询各大电商网站同一类产品的价格并汇总展示
    11.  */
    12. public class CompletableFutureUsage {
    13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    14.         way1();
    15.         way2();
    16.     }
    17.  
    18.     public static void way1() {
    19.         long start = System.currentTimeMillis();
    20.         System.out.println("p1 " + priceOfJD());
    21.         System.out.println("p2 " + priceOfTB());
    22.         System.out.println("p3 " + priceOfTM());
    23.         long end = System.currentTimeMillis();
    24.         System.out.println("串行执行，耗时(ms):" + (end - start));
    25.     }
    26.  
    27.     public static void way2() throws ExecutionException, InterruptedException {
    28.         long start = System.currentTimeMillis();
    29.         CompletableFuture p1 = CompletableFuture.supplyAsync(() -> priceOfJD());
    30.         CompletableFuture p2 = CompletableFuture.supplyAsync(() -> priceOfTB());
    31.         CompletableFuture p3 = CompletableFuture.supplyAsync(() -> priceOfTM());
    32.         CompletableFuture.allOf(p1, p2, p3).join();
    33.         System.out.println("p1 " + p1.get());
    34.         System.out.println("p2 " + p2.get());
    35.         System.out.println("p3 " + p3.get());
    36.         long end = System.currentTimeMillis();
    37.         System.out.println("使用CompletableFuture并行执行，耗时(ms): " + (end - start));
    38.     }
    39.  
    40.     private static double priceOfTM() {
    41.         delay();
    42.         return 1.00;
    43.     }
    44.  
    45.     private static double priceOfTB() {
    46.         delay();
    47.         return 2.00;
    48.     }
    49.  
    50.     private static double priceOfJD() {
    51.         delay();
    52.         return 3.00;
    53.     }
    54.  
    55.     private static void delay() {
    56.         int time = new Random().nextInt(500);
    57.         try {
    58.             TimeUnit.MILLISECONDS.sleep(time);
    59.         } catch (InterruptedException e) {
    60.             e.printStackTrace();
    61.         }
    62.     }
    63. }
    64.

    证明原子操作类比synchronized更高效

    1. package git.snippets.juc;
    2.  
    3. import java.util.ArrayList;
    4. import java.util.List;
    5. import java.util.concurrent.atomic.AtomicInteger;
    6.  
    7. /**
    8.  * 证明原子操作类比synchronized更高效
    9.  *
    10.  * @author Grey
    11.  * @date 2021/4/26
    12.  */
    13. public class AtomVSSync {
    14.     public static void main(String[] args) {
    15.         test1();
    16.     }
    17.  
    18.     AtomicInteger atomicCount = new AtomicInteger(0);
    19.     int count = 0;
    20.     final static int TIMES = 80000000;
    21.  
    22.     void m() {
    23.         for (int i = 0; i < TIMES; i++) {
    24.             atomicCount.incrementAndGet(); //原子操作
    25.         }
    26.     }
    27.  
    28.     void m2() {
    29.         for (int i = 0; i < TIMES; i++) {
    30.             synchronized (this) {
    31.                 count++;
    32.             }
    33.         }
    34.     }
    35.  
    36.  
    37.     public static void test1() {
    38.         AtomVSSync t1 = new AtomVSSync();
    39.         AtomVSSync t2 = new AtomVSSync();
    40.         long time1 = time(t1::m);
    41.         System.out.println("使用原子类得到的结果是：" + t1.atomicCount);
    42.         long time2 = time(t2::m2);
    43.         System.out.println("使用synchronized得到的结果是：" + t2.count);
    44.  
    45.         System.out.println("使用原子类花费的时间是：" + time1);
    46.         System.out.println("使用 synchronized 花费的时间是 ：" + time2);
    47.     }
    48.  
    49.     private static long time(Runnable runnable) {
    50.         List threads = new ArrayList<>();
    51.         long startTime = System.currentTimeMillis();
    52.         for (int i = 0; i < 10; i++) {
    53.             threads.add(new Thread(runnable, "thread-" + i));
    54.         }
    55.         threads.forEach(Thread::start);
    56.         threads.forEach(o -> {
    57.             try {
    58.                 o.join();
    59.             } catch (InterruptedException e) {
    60.                 e.printStackTrace();
    61.             }
    62.         });
    63.         long endTime = System.currentTimeMillis();
    64.         return endTime - startTime;
    65.     }
    66. }
    67.

  • 相关阅读:
    接口自动化Requests+Pytest基础实现
    EasyX趣味化编程note6，图片操作及文字
    Idea创建springboot项目
    使用 Promise.withResolvers() 来简化你将函数 Promise 化的实现~~
    2023-10-17 mysql-innodb-解析write_row的record的一行数据-分析
    flask——数据库连接池、wtfroms、信号、多app应用、flask-script
    里程碑，用自己的编程语言实现了一个网站
    python datetime模块
    C/C++语言的服务器LS调研 (Language Server 实现代码索引 跳转定义 智能提示等功能)
    Spring面试题25：Spring如何控制bean加载先后顺序
  • 原文地址：https://blog.csdn.net/chen978616649/article/details/126812828