【JUC】线程池_参数_ForkJoin_CompletableFuture

小鱼儿 2022-09-12 04:42 211阅读 0赞

# 线程池 #

类比为银行业务员，  
1、降低消耗，不用经常请人，hr也很幸苦的  
2、控制最大能有多少个业务员 MaxThreadPoolSize  
3、员工好管理，统一编制统一管理

**线程池：**  
第一：降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的销耗。  
第二：提高响应速度。当任务到达时，任务可以不需要等待线程创建就能立即执行。  
第三：提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会销耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

## 几个特殊的池 ##

**Executors.newFixedThreadPool(int)**：固定N个员工，可以根据当前CPU数量确定多少个线程  
**Executors.newSingleThreadExecutor()**：一个员工，但是等待池无限大  
**Executors.newCachedThreadPool()**：没有人自己就很弱，需求多自己线程就新建的多，根据压力来处理问题  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2lsbG9ybg_size_20_color_FFFFFF_t_70_g_se_x_16]

## 七大参数： ##

public ThreadPoolExecutor(int corePoolSize, 核心常驻线程大小
                            int maximumPoolSize, 最大线程大小
                            long keepAliveTime, 保持生存时间
                            TimeUnit unit, 时间单位
                            BlockingQueue<Runnable> workQueue, 等候区阻塞队列
                            ThreadFactory threadFactory, 线程工厂
                            RejectedExecutionHandler handler 拒绝策略) {

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2lsbG9ybg_size_20_color_FFFFFF_t_70_g_se_x_16 1]

【线程池工作原理】
     1、在创建了线程池后，开始等待请求。
     2、当调用execute()方法添加一个请求任务时，线程池会做出如下判断：  
    	2.1如果正在运行的线程数量小于corePoolSize，那么马上创建线程运行这个任务；  
    	2.2如果正在运行的线程数量大于或等于corePoolSize，那么将这个任务放入队列；  
    	2.3如果这个时候队列满了且正在运行的线程数量还小于maximumPoolSize，
    		那么还是要创建非核心线程立刻运行这个任务；  
    	2.4如果队列满了且正在运行的线程数量大于或等于maximumPoolSize，
    		那么线程池会启动饱和拒绝策略来执行。
     3、当一个线程完成任务时，它会从队列中取下一个任务来执行。
     4、当一个线程无事可做超过一定的时间（keepAliveTime）时，线程会判断：    
     	如果当前运行的线程数大于corePoolSize，那么这个线程就被停掉。    
     	所以线程池的所有任务完成后，它最终会收缩到corePoolSize的大小。

## 参数设置 ##

ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(3,
            5,
            2,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue(3),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy());  出厂默认

*  怎么确定**最大**线程数？
    
     *  CPU：1+Runtime.getRuntime().availableProcessors()
     *  IO集：1+因子

获取CPU数量：
    Runtime.getRuntime().availableProcessors()

### 拒绝策略 ###

1 等待队列已经排满了，再也塞不下新任务了  
2 线程池中的max线程也达到了，无法继续为新任务服务。  
》拒绝进入

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2lsbG9ybg_size_20_color_FFFFFF_t_70_g_se_x_16 2]

### JDK内置的拒绝策略 ###

【1】太多了直接报异常  
**AbortPolicy(默认)**：直接抛出RejectedExecutionException异常阻止系统正常运行

【2】不报错，将请求反弹回去给任务的提交者  
**CallerRunsPolicy**：“调用者运行”一种调节机制，该策略既不会抛弃任务，也不会抛出异常，而是将某些任务回退到调用者，从而降低新任务的流量。

【3】不报错，直接抛弃，允许丢失  
**DiscardOldestPolicy**：抛弃队列中等待最久的任务，然后把当前任务加人队列中尝试再次提交当前任务。  
**DiscardPolicy**：该策略默默地丢弃无法处理的任务，不予任何处理也不抛出异常。如果允许任务丢失，这是最好的一种策略。

## 分支合并框架 ForkJoinDemo ##

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2lsbG9ybg_size_20_color_FFFFFF_t_70_g_se_x_16 3]

### Demo ###

class RecursiveSource extends RecursiveTask<Integer> {
        
        private static final Integer hold = 10;
    
        private int begin;
        private int end;
        private int res;
    
        public RecursiveSource(int begin, int r) {
        
            this.begin = begin;
            this.end = r;
        }
    
        @Override
        protected Integer compute() {
        
            // baseCase：
            if (end - begin <= hold){
        
                for (int i = begin; i <= end; i++) {
        
                    res += i;
                }
            // 将问题fork出去
            }else{
        
                int mid = begin + ((end - begin) >> 1);
                RecursiveSource source1 = new RecursiveSource(begin, mid);
                RecursiveSource source2 = new RecursiveSource(mid+1, end);
                source1.fork();
                source2.fork();
                res = source1.join() + source2.join();
            }
            return res;
        }
    }
    
    class CallSource implements Callable<String> {
        
        @Override
        public String call() throws Exception {
        
            return "我现在很好，我是Callable接口";
        }
    }
    
    public class ForkDemo {
        
        public static void main(String[] args) throws ExecutionException, InterruptedException {
        
            //【1】
            /*将任务交给资源类处理*/
            /*资源类得到数据，呆住了，线程池进场*/
            /*使用submit代替线程的start方法*/
            RecursiveSource recursiveSource = new RecursiveSource(0,100);
            ForkJoinPool threadPool  = new ForkJoinPool();
            ForkJoinTask<Integer> forkJoinTask = threadPool.submit(recursiveSource);
            threadPool.shutdown();
    
            //【2】Callable接口对比，任务在线程中执行
            CallSource call = new CallSource();
            FutureTask<String> futureTask = new FutureTask<>(call);
            new Thread(futureTask).start();
    
            System.out.println(forkJoinTask.get());
            System.out.println(futureTask.get());
        }
    }

程序大概是这么个意思：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2lsbG9ybg_size_20_color_FFFFFF_t_70_g_se_x_16 4]

## 异步回调 CompletableFuture ##

隔夜问题：

public static void main(String[] args) throws Exception {
        
        //同步，异步，异步回调        //同步//
    //        CompletableFuture<Void> completableFuture1 = CompletableFuture.runAsync(() -> {//
    //            System.out.println(Thread.currentThread().getName() + "\t completableFuture1");//
    //        });
    //        completableFuture1.get();
    
        //异步回调
        CompletableFuture<Integer> completableFuture2 = CompletableFuture.supplyAsync(() -> {
        
            System.out.println(Thread.currentThread().getName() + "\t completableFuture2");
            int i = 10 / 0;
    
            return 1024;
        });
        completableFuture2.whenComplete((t, u) -> {
        
            System.out.println("-------t=" + t);
            System.out.println("-------u=" + u);
        }).exceptionally(f -> {
        
            System.out.println("-----exception:" + f.getMessage());
            return 444;
        }).get();
    }

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2lsbG9ybg_size_20_color_FFFFFF_t_70_g_se_x_16 5]

学习三板斧：理论，实操，小总结  
未完待续……

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2lsbG9ybg_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/b45a6f0d0dda4af185fc17cd64c727d0.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2lsbG9ybg_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/fb8937aa516f47a8b5a7baa025919efd.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2lsbG9ybg_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220828/548c92d0d12c44369b16191a5c07cf55.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2lsbG9ybg_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220828/e7def84c933048bc98ac4a6e4a51401c.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2lsbG9ybg_size_20_color_FFFFFF_t_70_g_se_x_16 4]: /images/20220828/c0c0cd003e9147e0ad9caa779f948381.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd2lsbG9ybg_size_20_color_FFFFFF_t_70_g_se_x_16 5]: /images/20220828/aa210af35d844dc08cfd0b7467e32913.png