并发编程CompletableFuture用法

CompletableFuture用法"/>

并发编程CompletableFuture用法

1.配置线程池

/*** int corePoolSize,* int maximumPoolSize,* long keepAliveTime,* TimeUnit unit,* BlockingQueue<Runnable> workQueue,* ThreadFactory threadFactory,* RejectedExecutionHandler handler** @return*/@Beanpublic ThreadPoolExecutor threadPoolExecutor(ThreadPoolConfigProperties pool) {return new ThreadPoolExecutor(pool.getCoreSize(),pool.getMaxSize(),pool.getKeepAliveTime(),TimeUnit.SECONDS,new LinkedBlockingQueue<>(100000),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());}

2.线程池参数配置类

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
/**my.thread.core-size=20my.thread.max-size=200my.thread.keep-alive-time=10
*/
@ConfigurationProperties(prefix = "my.thread")
@Component
@Data
public class ThreadPoolConfigProperties {//核心线程数private Integer coreSize;//最大线程数private Integer maxSize;//空余线程的存活时间private Integer keepAliveTime;
}

3.测试异步任务

	@Autowiredprivate ThreadPoolExecutor executor;//在这里开启一个异步任务，提交给线程池，runAsync()方法没有返回值，需要有返回值的可使用supplyAsync()方法@Testvoid testCompletableFuture() {CompletableFuture<Void> runAsync = CompletableFuture.runAsync(() -> {int result = 0;for (int i = 0; i <= 100; i++) {result += i;}System.out.println(result);}, executor);}

4.关于CompletableFuture的其他相关用法
4.1 CompletableFuture的**get()**方法可以获取异步的结果，get方法是一个阻塞式等待的方法，也即get方法会等待异步任务的完成

CompletableFuture<AtomicInteger> completableFuture2 = CompletableFuture.supplyAsync(() -> {for (int i = 0; i <= 100; i++) {sum2.addAndGet(i);}return sum2;}, executor);//获取异步结果AtomicInteger integer = completableFuture2.get();

4.2 allOf : 等待所有任务完成完成

AtomicInteger sum = new AtomicInteger();
AtomicInteger sum2 = new AtomicInteger();
CompletableFuture<AtomicInteger> completableFuture1 = CompletableFuture.supplyAsync(() -> {for (int i = 0; i <= 100; i++) {sum.addAndGet(i);}return sum;
}, executor);CompletableFuture<AtomicInteger> completableFuture2 = CompletableFuture.supplyAsync(() -> {for (int i = 0; i <= 100; i++) {sum2.addAndGet(i);}return sum2;
}, executor);
AtomicInteger integer = completableFuture2.get();//allOf : 等待所有任务完成完成，注意get方法，是阻塞式等待，等待上面的异步任务都完成
CompletableFuture.allOf(completableFuture1,completableFuture2).get();//获取异步结果AtomicInteger atomicInteger1 = completableFuture1.get();AtomicInteger atomicInteger2 = completableFuture2.get();System.out.println("结果是--->"+atomicInteger1.addAndGet(atomicInteger2.intValue()));

4.3 异步任务完成时，whenComplete，exceptionally

CompletableFuture<AtomicInteger> completableFuture3 = CompletableFuture.supplyAsync(() -> {for (int i = 0; i <= 10; i++) {sum2.addAndGet(i);}return sum2;}, executor).whenComplete((res, exception) -> {//当出现异常，可以拿到异常信息，但是无法修改返回数据System.out.println("结果是：" + res + ",异常：" + exception);}).exceptionally(throwable -> {//可以感知异常，同时返回默认值return new AtomicInteger(10);});

4.4 handle，方法完成后的后续处理

CompletableFuture<Integer> completableFuture4 = CompletableFuture.supplyAsync(() -> {int i = 10 / 2;return i;}, executor).handle((res, throwable) -> {//res 为结果，throwable 为异常if (res != null) {return res * 2;}if (throwable != null) {return -1;}return 0;});System.out.println("completableFuture4--结果是："+completableFuture4.get());

4.5 异步任务串行化

/**
* 异步任务串行化
*      thenAcceptAsync  可以接收上一步获取的结果，但是无返回值
*      thenApplyAsync   可以接收上一步获取的结果，有返回值
*/CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {int i = 10 / 2;return i;}, executor).thenApplyAsync(res -> {//res为上一步的结果return res * 2;}, executor).thenAcceptAsync((res) -> {System.out.println("hello ...thenAcceptAsync");}, executor);