IT数码 购物 网址 头条 软件 日历 阅读 图书馆
TxT小说阅读器
↓语音阅读,小说下载,古典文学↓ 		图片批量下载器
↓批量下载图片,美女图库↓ 		图片自动播放器
↓图片自动播放器↓ 		一键清除垃圾
↓轻轻一点,清除系统垃圾↓
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁
 
   -> Java知识库 -> JUC并发编程（三） --- 线程池，并行计算ForkJoin，异步回调FutureTask和CompletableFuture -> 正文阅读

[Java知识库]JUC并发编程（三） --- 线程池，并行计算ForkJoin，异步回调FutureTask和CompletableFuture

文章目录


JUC笔记

10、Callable回顾

详情：多线程拾遗

Callable与Runnable区别

  • 1、可以有返回值
  • 2、可以抛出异常
  • 3、方法不同，run()/ call()
public class CallableDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // new Thread(new Runnable()).start();
        // new Thread(new FutureTask<V>()).start();             FutureTask放进Thread里  和 Runnable放进Thread等价
        // new Thread(new FutureTask<V>( Callable )).start();   callable放进FutureTask里

        MyThread myThread = new MyThread();
        FutureTask<String> task = new FutureTask<>(myThread);
        new Thread(task,"A").start();
        new Thread(task,"B").start();

        String s = task.get();
        System.out.println(s);

    }
    static class MyThread implements Callable<String> {
        @Override
        public String call() {
            System.out.println("call()"); // 会打印几个call
            return Thread.currentThread().getName()+"->call";
        }
    }
}

11、线程池

池化技术：事先准备一些资源，需要了就来拿，用完了就还给我。

线程池好处：

  1. 降低资源消耗
  2. 提高响应速度
  3. 方便管理

线程复用，控制最大并发数，管理线程

重点：三大方法、7大参数、4种拒绝策略

11.1 三大方法

三大方法指Executors工具类里创建线程池的三大方法

  • Executors.newSingleThreadExecutor()：创建单个线程池
  • Executors.newFixedThreadPool(int nums)：创建一个固定大小为nums的线程池
  • Executors.newCachedThreadPool()：创建一个可伸缩的线程池，遇强则强，遇弱则弱
public class ThreadPoolDemo {
    public static void main(String[] args) {
        ExecutorService threadPool = Executors.newSingleThreadExecutor();//单个线程池
//        ExecutorService threadPool = Executors.newFixedThreadPool(5);//一个固定大小为5的线程池
//        ExecutorService threadPool = Executors.newCachedThreadPool();//创建一个可伸缩的线程池

        try {
            for (int i = 0; i < 100; i++) {
                //用execute来创建一个线程
                threadPool.execute(()->{
                    System.out.println(Thread.currentThread().getName()+"created");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //用完了要记得关闭
            threadPool.shutdown();
        }
    }
}

11.2 七大参数

先分析一波源码

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService(
        new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
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>());
}

可以看到这三个方法本质是调用的ThreadPoolExecutor()，这个方法的7个参数十分重要

public ThreadPoolExecutor(int corePoolSize,		// 核心线程池大小
                          int maximumPoolSize, 	// 最大核心线程池大小
                          long keepAliveTime,	// 超时了多久没有人调用就会释放
                          TimeUnit unit,		// 超时的单位：SECOND，MINUTE...
                          BlockingQueue<Runnable> workQueue,	//阻塞队列
                          ThreadFactory threadFactory,			// 线程工厂：用来创建线程的，一般不用动
                          RejectedExecutionHandler handler) {	//拒绝策略：4个拒绝策略
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.acc = System.getSecurityManager() == null ?
        null :
    AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

如图：上面几个参数的形象描述

image-20220107181457036

注意注意注意！！！

image-20220107181212363

阿里巴巴开发手册里明确了线程池不能使用Executors的三个方法进行创建，而要用ThreadPoolExecutor原始方法创建

很大的弊端：如图，会造成OOM

11.3 四个拒绝策略

  • new ThreadPoolExecutor.AbortPolicy() // 银行满了，还有人进来，不处理这个人的，抛出异常
  • new ThreadPoolExecutor.CallerRunsPolicy() // 哪来的去哪里！
  • new ThreadPoolExecutor.DiscardPolicy() // 队列满了，丢掉任务，不会抛出异常！
  • new ThreadPoolExecutor.DiscardOldestPolicy() // 队列满了，尝试去和最早的竞争，也不会抛出异常！

现在来自定义一个线程池：

public class ThreadPoolDemo02 {

    public static void main(String[] args) {
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 5, 3, TimeUnit.SECONDS,
                                                                new ArrayBlockingQueue<>(3), 
                                                                Executors.defaultThreadFactory(), 
                                                                new ThreadPoolExecutor.DiscardPolicy());//队列满了，尝试去和
最早的竞争，也不会抛出异常！
        try {
            for (int i = 0; i < 9; i++) {
                threadPool.execute(()->{
                    System.out.println(Thread.currentThread().getName()+" used");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool.shutdown();
        }
    }

}

11.4 池的大小怎么确定

判断IO密集型，CPU密集型（池的调优）

  • 1、CPU 密集型，几核，就是几，可以保持CPU的效率最高。
    • 设置公式：线程数 = CPU 核心数 + 1 （超线程）
  • 2、IO 密集型 ， 判断你程序中十分耗IO的线程
  • 设置公式：线程数 = （ 1 + 线程等待时间 / 线程 CPU 时间 ）* CPU 核数 * CPU 使用率
  • 或者简单粗暴：线程数 = 2 * CPU 核数

image-20220107194309376

Runtime.getRuntime().availableProcessors()获得核数

public class ThreadPoolDemo02 {

    public static void main(String[] args) {
        System.out.println(Runtime.getRuntime().availableProcessors());

        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2,
                                                                Runtime.getRuntime().availableProcessors(),//此处最大设置为cpu的核数
                                                                3, TimeUnit.SECONDS,
                                                                new ArrayBlockingQueue<>(3),
                                                                Executors.defaultThreadFactory(),
                                                                new ThreadPoolExecutor.DiscardPolicy());
        try {
            for (int i = 0; i < 9; i++) {
                threadPool.execute(()->{
                    System.out.println(Thread.currentThread().getName()+" used");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool.shutdown();
        }
    }

}

12、并行计算ForkJoin

12.1 什么是ForkJoin

ForkJoin 在 JDK 1.7出现 ， 用来并行执行任务！从而提高效率，处理大数据量

image-20220107191912746

12.2 ForkJoin特点

工作窃取：

如上图，ForkJoin维护的是双端队列，当一条队列执行完之后，可以窃取其他队列的任务来协助执行

12.3 ForkJoinTask和ForkJoinPool

ForkJoinTask

image-20220107192632146 实际上ForkJoinTask是轻量级的FutureTask

ForkJoinPool

image-20220107192418024

ForkJoinPool用来执行ForkJoinTask

实践：

要使用ForkJoin，我们继承一个ForkJoinTask的实现类，需要重写compute方法

public class ForkJoinDemo extends RecursiveTask<Long> {
    @Override
    protected Long compute() {
        return null;
    }
}

/**
 * 求和计算的任务！
 * 3000 6000（ForkJoin） 9000（Stream并行流）
 * // 如何使用 forkjoin
 * // 1、forkjoinPool 通过它来执行
 * // 2、计算任务 forkjoinPool.execute(ForkJoinTask task)
 * // 3. 计算类要继承 ForkJoinTask
 */
public class ForkJoinDemo extends RecursiveTask<Long> {

    private Long start; // 1
    private Long end; // 1990900000
    // 临界值
    private Long temp = 10000L;
    public ForkJoinDemo(Long start, Long end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        if ((end - start) < temp) {
            Long sum = 0L;
            for (Long i = start; i <= end; i++) {
                sum += i;
            }
            return sum;
        } else { // forkjoin 递归
            long middle = (start + end) / 2; // 中间值
            ForkJoinDemo task1 = new ForkJoinDemo(start, middle);
            task1.fork(); // 拆分任务，把任务压入线程队列
            ForkJoinDemo task2 = new ForkJoinDemo(middle + 1, end);
            task2.fork(); // 拆分任务，把任务压入线程队列
            return task1.join() + task2.join();
        }
    }
}

public class Test {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        test1();    //计算需要  5509
        test2();    //计算需要  3766    
        test3();    //计算需要  203     流式计算 yyds
    }


    // 普通程序员
    public static void test1(){
        Long sum = 0L;
        long start = System.currentTimeMillis();
        for (Long i = 1L; i <= 10_0000_0000; i++) {
            sum += i;
        }
        long end = System.currentTimeMillis();
        System.out.println("sum="+sum+" 时间："+(end-start));
    }
    // 会使用ForkJoin
    public static void test2() throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Long> task = new ForkJoinDemo(0L, 10_0000_0000L);
        ForkJoinTask<Long> submit = forkJoinPool.submit(task);// 提交任务
        Long sum = submit.get();
        long end = System.currentTimeMillis();
        System.out.println("sum="+sum+" 时间："+(end-start));
    }

    public static void test3(){
        long start = System.currentTimeMillis();
        // Stream并行流 ()
        long sum = LongStream.rangeClosed(0L,
                10_0000_0000L).parallel().reduce(0, Long::sum);
        long end = System.currentTimeMillis();
        System.out.println("sum="+"时间："+(end-start));
    }
}

13、异步回调Future和FutureTask

13.1 Future

来看看jdk文档

image-20220107195026512

Future作为一个接口，对具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。

必要时可以通过get方法获取执行结果，该方法会阻塞直到任务返回结果。

其实现类有ForkJoinTask（上面已经提到过），FutureTask（接下来聊聊FutureTask）

13.2 FutureTask

查看jdk文档介绍：

该类提供了一个Future的基本实现 ，具有启动和取消计算的方法，查询计算是否完整，并检索计算结果。 结果只能在计算完成后才能检索; 如果计算尚未完成，则get方法将阻止。 一旦计算完成，则无法重新启动或取消计算（除非使用runAndReset()调用计算）。

FutureTask分为三种状态：

  • 未启动。FutureTask.run()方法还没有被执行之前，FutureTask处于未启动状态。当创建一个FutureTask，还没有执行FutureTask.run()方法之前，FutureTask处于未启动状态。

  • 已启动。FutureTask.run()方法被执行的过程中，FutureTask处于已启动状态。

  • 已完成。FutureTask.run()方法执行结束，或者调用FutureTask.cancel(…)方法取消任务，或者在执行任务期间抛出异常，这些情况都称之为FutureTask的已完成状态。

image-20220107200537763

Demo：

public class CallableDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        MyThread myThread = new MyThread();
        FutureTask<String> task = new FutureTask<>(myThread);
        new Thread(task,"A").start();
        new Thread(task,"B").start();

        String s = task.get();
        System.out.println(s);

    }
    static class MyThread implements Callable<String> {
        @Override
        public String call() {
            System.out.println("call()"); // 会打印几个call  1个，futureTask执行一次就改变状态了，不能再次执行
            return Thread.currentThread().getName()+"->call";
        }
    }
}

13.3 CompletableFuture

image-20220107225233084

FutureTask获取异步任务执行结果时，要么调用get阻塞，要么轮询isDone主动查询是否完成。这两种方法都不太好，会使主线程被迫等待。

而CompletableFuture完成一个任务阶段后可以主动通知下一个阶段

简单的使用：

  • runAsync：没有返回值的回调
  • supplyAsync：有返回值的回调
public class CompletableFutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        test1();
        test2();
    }

    public static void test1() throws ExecutionException, InterruptedException {
         //没有返回值的 runAsync 异步回调
         CompletableFuture<Void> completableFuture =CompletableFuture.runAsync(()->{
             try {
                TimeUnit.SECONDS.sleep(2);
             } catch (InterruptedException e) {
                e.printStackTrace();
             }
            System.out.println(Thread.currentThread().getName()+"runAsync=>Void");
         });
         System.out.println("1111");
         completableFuture.get(); // 获取阻塞执行结果
    }

    public static void test2() throws ExecutionException, InterruptedException {
        // 有返回值的 supplyAsync 异步回调
        // ajax，成功和失败的回调
        // 返回的是错误信息；
        CompletableFuture<Integer> completableFuture =
                CompletableFuture.supplyAsync(()->{
                    System.out.println(Thread.currentThread().getName()+"supplyAsync=>Integer");
                    int i = 10/0;
                    return 1024;
                });
        System.out.println(completableFuture.whenComplete((t, u) -> {
            System.out.println("t=>" + t); // 正常的返回结果
            System.out.println("u=>" + u); // 错误信息：java.util.concurrent.CompletionException: java.lang.ArithmeticException: / byzero
        }).exceptionally((e) -> {
            System.out.println(e.getMessage());
            return 233; // 可以获取到错误的返回结果
        }).get());
    }
}
  Java知识库 最新文章
计算距离春节还有多长时间
系统开发系列 之WebService（spring框架+ma
springBoot+Cache(自定义有效时间配置)
SpringBoot整合mybatis实现增删改查、分页查
spring教程
SpringBoot+Vue实现美食交流网站的设计与实
虚拟机内存结构以及虚拟机中销毁和新建对象
SpringMVC---原理
小李同学: Java如何按多个字段分组
打印票据--java
上一篇文章      下一篇文章      查看所有文章
加:2022-01-08 13:51:00  更:2022-01-08 13:54:19 
 
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁

360图书馆 购物 三丰科技 阅读网 日历 万年历 2024年11日历 -2024/11/24 9:09:41-

图片自动播放器
↓图片自动播放器↓ 		TxT小说阅读器
↓语音阅读,小说下载,古典文学↓ 		一键清除垃圾
↓轻轻一点,清除系统垃圾↓ 		图片批量下载器
↓批量下载图片,美女图库↓
  网站联系: qq:121756557 email:121756557@qq.com  IT数码