资讯详情

谷粒商城-10-p193-p247

193、商城业务-异步-异步复习

百度的线程基础

异步编排可参考网上链接：https://blog.csdn.net/weixin_45762031/article/details/103519459

线程回顾

初始化线程的四种方式：

1、继承Thread 2、实现Runnable接口 3、实现Callable接口 FutureTask (可获得返回结果，可处理异常) 4、线程池 方法1和方法2：主过程无法获得线程的计算结果。不适合当前场景

方法3：主过程可获得线程的计算结果，并设置为itemVO，但不利于控制服务器中的线程资源。可导致服务器资源耗尽。

初始化线程池有以下两种方式：

Executors.newFiexedThreadPool(3); //或者 new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit unit, workQueue, threadFactory, handler);

执行结果也可以通过稳定的线程池获得并捕获异常。然而，在业务复杂的情况下，一个异步调用可能取决于另一个异步调用的执行结果。

public class ThreadTest { 
             public static ExecutorService executor = Executors.newFixedThreadPool(10);       public static void main(String[] args) throws ExecutionException, InterruptedException { 
                 System.out.println("main...start..."); // CompletableFuture<Void> future = CompletableFuture.runAsync(() -> { 
         // System.out.println("当前线程：" Thread.currentThread().getId()); // int i = 10 / 2; // System.out.println("运行结果：" i); // }, executor);          /** * 方法完成后的感知 */ // CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> { 
         // System.out.println("当前线程：" Thread.currentThread().getId()); // int i = 10 / 0; // System.out.println("运行结果：" i); // return i; // }, executor).whenComplete((res,excption)->{ 
         // //虽然可以获得异常信息，但返回数据无法修改。 // System.out.println("异步任务成功完成...结果是：" res ";异常是:" excption); // }).exceptionally(throwable -> { 
         // //可以感知异常，同时返回默认值 // return 10; // });           /** * 方法实施后的处理 */ // CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> { 
          // System.out.println("当前线程：" + Thread.currentThread().getId()); // int i = 10 / 4; // System.out.println("运行结果：" + i); // return i; // }, executor).handle((res, thr) -> { 
          // if (res != null) { 
          // return res * 2; // } // if (thr != null) { 
          // return 0; // } // return 0; // }); //R apply(T t, U u); /** * 线程串行化 * 1）、thenRun：不能获取到上一步的执行结果，无返回值 * .thenRunAsync(() -> { * System.out.println("任务2启动了..."); * }, executor); * 2）、thenAcceptAsync;能接受上一步结果，但是无返回值 * 3）、thenApplyAsync：;能接受上一步结果，有返回值 */ // CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { 
          // System.out.println("当前线程：" + Thread.currentThread().getId()); // int i = 10 / 4; // System.out.println("运行结果：" + i); // return i; // }, executor).thenApplyAsync(res -> { 
          // System.out.println("任务2启动了..." + res); // // return "Hello " + res; // }, executor); //void accept(T t); //R apply(T t); //future.get() /** * 两个都完成 */ // CompletableFuture<Object> future01 = CompletableFuture.supplyAsync(() -> { 
          // System.out.println("任务1线程：" + Thread.currentThread().getId()); // int i = 10 / 4; // System.out.println("任务1结束：" ); // return i; // }, executor); // // CompletableFuture<Object> future02 = CompletableFuture.supplyAsync(() -> { 
          // System.out.println("任务2线程：" + Thread.currentThread().getId()); // // try { 
          // Thread.sleep(3000); // System.out.println("任务2结束：" ); // } catch (InterruptedException e) { 
          // e.printStackTrace(); // } // return "Hello"; // }, executor); // future01.runAfterBothAsync(future02,()->{ 
          // System.out.println("任务3开始..."); // },executor); // void accept(T t, U u); // future01.thenAcceptBothAsync(future02,(f1,f2)->{ 
          // System.out.println("任务3开始...之前的结果："+f1+"--》"+f2); // },executor); //R apply(T t, U u); // CompletableFuture<String> future = future01.thenCombineAsync(future02, (f1, f2) -> { 
          // return f1 + "：" + f2 + " -> Haha"; // }, executor); /** * 两个任务，只要有一个完成，我们就执行任务3 * runAfterEitherAsync：不感知结果，自己没有返回值 * acceptEitherAsync：感知结果，自己没有返回值 * applyToEitherAsync：感知结果，自己有返回值 */ // future01.runAfterEitherAsync(future02,()->{ 
          // System.out.println("任务3开始...之前的结果："); // },executor); //void accept(T t); // future01.acceptEitherAsync(future02,(res)->{ 
          // System.out.println("任务3开始...之前的结果："+res); // },executor); // CompletableFuture<String> future = future01.applyToEitherAsync(future02, res -> { 
          // System.out.println("任务3开始...之前的结果：" + res); // return res.toString() + "->哈哈"; // }, executor); CompletableFuture<String> futureImg = CompletableFuture.supplyAsync(() -> { 
          System.out.println("查询商品的图片信息"); return "hello.jpg"; },executor); CompletableFuture<String> futureAttr = CompletableFuture.supplyAsync(() -> { 
          System.out.println("查询商品的属性"); return "黑色+256G"; },executor); CompletableFuture<String> futureDesc = CompletableFuture.supplyAsync(() -> { 
          try { 
          Thread.sleep(3000); System.out.println("查询商品介绍"); } catch (InterruptedException e) { 
          e.printStackTrace(); } return "华为"; },executor); // CompletableFuture<Void> allOf = CompletableFuture.allOf(futureImg, futureAttr, futureDesc); CompletableFuture<Object> anyOf = CompletableFuture.anyOf(futureImg, futureAttr, futureDesc); anyOf.get();//等待所有结果完成 // System.out.println("main....end...."+futureImg.get()+"=>"+futureAttr.get()+"=>"+futureDesc.get()); System.out.println("main....end...."+anyOf.get()); } public void thread(String[] args) throws ExecutionException, InterruptedException { 
          System.out.println("main....start...."); /** * 1）、继承Thread * Thread01 thread = new Thread01(); * thread.start();//启动线程 * * 2）、实现Runnable接口 * Runable01 runable01 = new Runable01(); * new Thread(runable01).start(); * 3）、实现Callable接口 + FutureTask （可以拿到返回结果，可以处理异常） * FutureTask<Integer> futureTask = new FutureTask<>(new Callable01()); * new Thread(futureTask).start(); * //阻塞等待整个线程执行完成，获取返回结果 * Integer integer = futureTask.get(); * 4）、线程池[ExecutorService] * 给线程池直接提交任务。 * service.execute(new Runable01()); * 1、创建： * 1）、Executors * 2）、new ThreadPoolExecutor * * Future:可以获取到异步结果 * * 区别; * 1、2不能得到返回值。3可以获取返回值 * 1、2、3都不能控制资源 * 4可以控制资源，性能稳定。 */ //我们以后再业务代码里面，以上三种启动线程的方式都不用。【将所有的多线程异步任务都交给线程池执行】 // new Thread(()-> System.out.println("hello")).start(); //当前系统中池只有一两个，每个异步任务，提交给线程池让他自己去执行就行 /** * 七大参数 * corePoolSize:[5] 核心线程数[一直存在除非（allowCoreThreadTimeOut）]; 线程池，创建好以后就准备就绪的线程数量，就等待来接受异步任务去执行。 * 5个 Thread thread = new Thread(); thread.start(); * maximumPoolSize:[200] 最大线程数量; 控制资源 * keepAliveTime:存活时间。如果当前的线程数量大于core数量。 * 释放空闲的线程（maximumPoolSize-corePoolSize）。只要线程空闲大于指定的keepAliveTime； * unit:时间单位 * BlockingQueue<Runnable> workQueue:阻塞队列。如果任务有很多，就会将目前多的任务放在队列里面。 * 只要有线程空闲，就会去队列里面取出新的任务继续执行。 * threadFactory:线程的创建工厂。 * RejectedExecutionHandler handler:如果队列满了，按照我们指定的拒绝策略拒绝执行任务 * * * * 工作顺序: * 1)、线程池创建，准备好core数量的核心线程，准备接受任务 * 1.1、core满了，就将再进来的任务放入阻塞队列中。空闲的core就会自己去阻塞队列获取任务执行 * 1.2、阻塞队列满了，就直接开新线程执行，最大只能开到max指定的数量 * 1.3、max满了就用RejectedExecutionHandler拒绝任务 * 1.4、max都执行完成，有很多空闲.在指定的时间keepAliveTime以后，释放max-core这些线程 * * new LinkedBlockingDeque<>()：默认是Integer的最大值。内存不够 * * 一个线程池 core 7； max 20 ，queue：50，100并发进来怎么分配的； * 7个会立即得到执行，50个会进入队列，再开13个进行执行。剩下的30个就使用拒绝策略。 * 如果不想抛弃还要执行。CallerRunsPolicy； * */ ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 200, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(100000), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy()); // Executors.newCachedThreadPool() core是0，所有都可回收 // Executors.newFixedThreadPool() 固定大小，core=max；都不可回收 // Executors.newScheduledThreadPool() 定时任务的线程池 // Executors.newSingleThreadExecutor() 单线程的线程池，后台从队列里面获取任务，挨个执行 // System.out.println("main....end...."); } public static class Thread01 extends Thread{ 
          @Override public void run() { 
          System.out.println("当前线程："+Thread.currentThread().getId()); int i = 10 / 2; System.out.println("运行结果："+i); } } public static class Runable01 implements Runnable{ 
          @Override public void run() { 
          System.out.println("当前线程："+Thread.currentThread().getId()); int i = 10 / 2; System.out.println("运行结果："+i); } } public static class Callable01 implements Callable<Integer>{ 
          @Override public Integer call() throws Exception { 
          System.out.println("当前线程："+Thread.currentThread().getId()); int i = 10 / 2; System.out.println("运行结果："+i); return i; } } }

194、商城业务-异步-线程池详解

### 线程池的七大参数

corePoolSize 池中一直保持的线程的数量，即使线程空闲。除非设置了allowCoreThreadTimeOut

maximumPoolSize 池中允许的最大的线程数

keepAliveTime 当线程数大于核心线程数的时候，线程在最大多长时间没有接到新任务就会终止释放，
最终线程池维持在corePoolSize大小

unit 时间单位

workQueue 阻塞队列，用来存储等待执行的任务，如果当前对线程的需求超过了corePoolSize大小，就会放在这里等待空闲线程执行。

threadFactory 创建线程的工厂，比如指定线程名等
handler 拒绝策略，如果线程满了，线程池就会使用拒绝策略。



# 运行流程：

1、线程池创建，准备好 core 数量的核心线程，准备接受任务

2、新的任务进来，用 core 准备好的空闲线程执行。

(1)   、core 满了，就将再进来的任务放入阻塞队列中。空闲的 core 就会自己去阻塞队列获取任务执行

(2)   、阻塞队列满了，就直接开新线程执行，最大只能开到 max 指定的数量

(3)   、max 都执行好了。Max-core 数量空闲的线程会在 keepAliveTime 指定的时间后自动销毁。最终保持到 core 大小

(4)   、如果线程数开到了 max 的数量，还有新任务进来，就会使用 reject 指定的拒绝策略进行处理

3、所有的线程创建都是由指定的 factory 创建的。

面试：

一个线程池 core 7； max 20 ，queue：50，100 并发进来怎么分配的；先有 7 个能直接得到执行，接下来 50 个进入队列排队，在多开 13 个继续执行。现在 70 个被安排上了。剩下 30 个默认拒绝策略。

### 3、常见的 4 种线程池

1、 newCachedThreadPool

 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。

2、newFixedThreadPool

 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。

3、newScheduledThreadPool

创建一个定长线程池，支持定时及周期性任务执行。

4、newSingleThreadExecutor

创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。


### 4、开发中为什么使用线程池

l 、降低资源的消耗

 通过重复利用已经创建好的线程降低线程的创建和销毁带来的损耗

2、提高响应速度

 因为线程池中的线程数没有超过线程池的最大上限时，有的线程处于等待分配任务的状态，当任务来时无需创建新的线程就能执行

3、 提高线程的可管理性

 线程池会根据当前系统特点对池内的线程进行优化处理，减少创建和销毁线程带来的系统开销。无限的创建和销毁线程不仅消耗系统资源，还降低系统的稳定性，使用线程池进行统一分配

线程池处理jdk提供的以外，spring也提供了线程池的封装 具体查看下面的连接

spring 线程池和java线程池 - 简书 (jianshu.com) https://www.jianshu.com/p/77720cb0fdfb

195、商城业务-异步-CompletableFuture

业务场景：

查询商品详情页的逻辑比较复杂，有些数据还需要远程调用，必然需要花费更多的时间。

image-20220501221223080

假如商品详情页的每个查询，需要如下标注的时间才能完成那么，用户需要 5.5s 后才能看到商品详情页的内容。很显然是不能接受的。 如果有多个线程同时完成这 6 步操作，也许只需要 1.5s 即可完成响应。

Future 是 Java 5 添加的类，用来描述一个异步计算的结果。你可以使用isDone方法检查计算是否完成，或者使用get阻塞住调用线程，直到计算完成返回结果，你也可以使用cancel方法停止任务的执行。 虽然Future以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不 方便，只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然和我们的异步编程的初衷相违背，轮询的方式又会耗费无谓的 CPU 资源，而且也不能及时地得到计算结果，为什么不能用观察者设计模式当计算结果完成及时通知监听者呢？ 很多语言，比如 Node.js，采用回调的方式实现异步编程。Java 的一些框架，比如 Netty，自己扩展了 Java 的 Future接口，提供了addListener等多个扩展方法；Google guava 也提供了通用的扩展 Future；Scala 也提供了简单易用且功能强大的 Future/Promise 异步编程模式。 作为正统的 Java 类库，是不是应该做点什么，加强一下自身库的功能呢？在 Java 8 中, 新增加了一个包含 50 个方法左右的类: CompletableFuture，提供了非常强大的Future 的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以 通过回调的方式处理计算结果，并且提供了转换和组合 CompletableFuture 的方法。 CompletableFuture 类实现了 Future 接口，所以你还是可以像以前一样通过get方法阻塞或 者轮询的方式获得结果，但是这种方式不推荐使用。CompletableFuture 和 FutureTask 同属于 Future 接口的实现类，都可以获取线程的执行结果。

196、商城业务-异步-CompletableFuture-启动异步任务

CompletableFuture 提供了四个静态方法来创建一个异步操作。

static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

1、runXxxx 都是没有返回结果的，supplyXxx 都是可以获取返回结果的

2、可以传入自定义的线程池，否则就用默认的线程池；

没有指定Executor的方法会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。如果指定线程池，则使用指定的线程池运行。以下所有的方法都类同。runAsync方法不支持返回值。runXxxx 都是没有返回结果的，supplyXxx 都是可以获取返回结果的 supplyAsync可以支持返回值。

2、编码实现

public class ThreadTest { 
        

    public static ExecutorService executor = Executors.newFixedThreadPool(10);
    public static void main(String[] args) throws ExecutionException, InterruptedException { 
        
        System.out.println("main....start....");
      
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> { 
        
            System.out.println("当前线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
        }, executor);
      
      
       CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> { 
        
            System.out.println("当前线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
            return i;
        }, executor);

        System.out.println(future1.get());
    }
}

197、商城业务-异步-CompletableFuture-完成回调与异常感知

1、

public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action);
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action);
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor);

public CompletableFuture<T> exceptionally(Function<Throwable,? extends T> fn);

whenComplete 可以处理正常和异常的计算结果，exceptionally 处理异常情况。

whenComplete 和 whenCompleteAsync 的区别：

whenComplete：是执行当前任务的线程执行继续执行 whenComplete 的任务。

whenCompleteAsync：是执行把 whenCompleteAsync 这个任务继续提交给线程池来进行执行。

   		 /** * 方法完成后的感知 */
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> { 
        
            System.out.println("当前线程：" + Thread.currentThread().getId());
            int i = 10 / 0;
            System.out.println("运行结果：" + i);
            return i;
        }, executor).whenComplete((res,excption)->{ 
        
            //虽然能得到异常信息，但是没法修改返回数据。
            System.out.println("异步任务成功完成了...结果是："+res+";异常是:"+excption);
        }).exceptionally(throwable -> { 
        
            //可以感知异常，同时返回默认值
            return 10;
        });

System.out.println(future.get());

案列二

public class CompletableFutureDemo { 
        

    public static void main(String[] args) throws ExecutionException, InterruptedException { 
        
        CompletableFuture future = CompletableFuture.supplyAsync(new Supplier<Object>() { 
        
            @Override
            public Object get() { 
        
                System.out.println(Thread.currentThread().getName() + "\t completableFuture");
                int i = 10 / 0;
                return 1024;
            }
        }).whenComplete(new BiConsumer<Object, Throwable>() { 
        
            @Override
            public void accept(Object o, Throwable throwable) { 
        
                System.out.println("-------o=" + o.toString());
                System.out.println("-------throwable=" + throwable);
            }
        }).exceptionally(new Function<Throwable, Object>() { 
        
            @Override
            public Object apply(Throwable throwable) { 
        
                System.out.println("throwable=" + throwable);
                return 6666;
            }
        });
        System.out.println(future.get());
    }
}

198、商城业务-异步-CompletableFuture-handle最终处理

handle 是执行对结果的处理。 handle 是在任务完成后再执行，还可以处理异常的任务。

public <U> CompletionStage<U 

        标签： 直流电流传感器tfy
 
锐单商城拥有海量元器件数据手册、
          IC替代型号，打造
          电子元器件IC百科大全！

锐单商城 - 一站式电子元器件采购平台