文章目录
- 异步线程池讲解
- 简单线程池
- 常见的四种线程池
- 进阶线程池
- 为什么使用线程池
- 异步编排
- 基本用法
- 其他API
- 线程串行化
- 两任务组合
- 都完成时
- 一个完成时
- 多任务组合
异步线程池讲解
简单线程池
public class Test01 {
public static void main(String[] args) {
// 声明一个有10个线程的线程池
ExecutorService executorService = Executors.newFixedThreadPool(10);
// 传入一个Runnable接口,并执行run()方法
executorService.execute(new Runnable() {
@Override
public void run() {
while (true){
System.out.println("线程池内的输出:A");
}
}
});
while (true){
System.out.println("线程池外的输出:B");
}
}
}
可以看到,控制台输出中,既有A也有B,故实现了异步执行的效果:
常见的四种线程池
进阶线程池
public class Test02 {
public static void main(String[] args) {
// 核心线程数,会一直占用在线程池中
int corePoolSize = 5;
// 最大线程数(控制资源)
int maximumPoolSize = 200;
// 存活时间(单位与unit一致,类型为long)。如果当前线程数大于核心线程数
long keepAliveTime = 2000L;
// unit:时间单位 这里使用毫秒
// MILLISECONDS(毫秒),SECONDS(秒),MINUTES(分钟),HOURS(小时),DAYS(天)
TimeUnit timeUnit = TimeUnit.MILLISECONDS;
// 阻塞队列(如果没有空闲线程,且当前线程已经达到max,则进入阻塞队列等待)
BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<Runnable>(100);
// threadFactory: 线程的创建工厂
ThreadFactory defaultThreadFactory = Executors.defaultThreadFactory();
// handler: 如果阻塞队列满了,则按照我们指定的拒绝策略执行任务
// 自带的几种拒绝策略:
// 1. AbortPolicy:当任务添加到线程池中被拒绝时,它将抛出 RejectedExecutionException 异常(拒绝新来的任务,并抛出异常)
// 2. CallerRunsPolicy:当任务添加到线程池中被拒绝时,会在线程池当前正在运行的Thread线程池中处理被拒绝的任务。(直接允许Run方法,同步调用)
// 3. DiscardPolicy:当任务添加到线程池中被拒绝时,线程池将丢弃被拒绝的任务。(拒绝新来的任务,但不抛异常)
// 4. DiscardOldestPolicy:当任务添加到线程池中被拒绝时,线程池会放弃等待队列中最旧的未处理任务,然后将被拒绝的任务添加到等待队列中。(为接收新任务,丢弃最老的任务)
ThreadPoolExecutor.AbortPolicy abortPolicy = new ThreadPoolExecutor.AbortPolicy();
// 创建线程池
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
corePoolSize,
maximumPoolSize,
keepAliveTime,
timeUnit,
blockingQueue,
defaultThreadFactory,
abortPolicy);
// 使用线程池
threadPoolExecutor.execute(new Runnable() {
@Override
public void run() {
while (true){
System.out.println("线程池内的输出:A");
}
}
});
while (true){
System.out.println("线程池外的输出:B");
}
}
}
也可以实现同样的效果,但是比简单的线程池更好的控制了线程资源的使用
为什么使用线程池
异步编排
基本用法
先演示一下正常情况下,CompletableFuture异步编排的效果
public static void main(String[] args) throws ExecutionException, InterruptedException {
// 首先创建一个线程池,创建方法参照之前的代码,这里直接封装成方法
ThreadPoolExecutor threadPoolExecutor = createExector();
// 开始异步编排
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
// 异步任务,相当于run()方法
System.out.println("当前线程名:" + Thread.currentThread().getName());
int i = 10 / 2;
System.out.println("运行结果:" + i);
return i;
}, threadPoolExecutor).whenComplete((result, exception) -> {
// 如果上面的异步任务执行完毕了,就进入这里,传入参数是上面任务的结果result和异常信息exception
System.out.println("whenComplete中 :异步任务完成了,结果是:" + result + ",异常是:" + exception);
}).exceptionally(exception -> {
// 如果发生异常就进入这里,在这里可以对返回结果进行修正,或者手动抛出异常 exception.printStackTrace()
System.out.println("exceptionally中:发生了异常:" + exception);
return 10;
});
// 获取异步任务结果
Integer result = future.get();
System.out.println("main...end..."+result);
}
有异常的情况,将supplyAsync中的代码修改为如下代码:
// 异步任务,相当于run()方法
System.out.println("当前线程名:" + Thread.currentThread().getName());
int i = 10 / 0;
System.out.println("运行结果:" + i);
return i;
总结:
- 首先执行supplyAsync中的代码
- supplyAsync中代码执行完毕后(抛出异常也算执行完毕)就进入whenComplete中,whenComplete会获取supplyAsync的结果信息和异常信息
- 如果supplyAsync中的代码有异常,则会在whenComplete中的代码执行完之后,进入exceptionally中,exceptionally会获取异常信息,并可以决定是否手动抛出异常 exception.printStackTrace(),还可以手动返回一个默认结果(类似服务降级机制)
其他API
handle方法相当于结合了whenComplete和exceptionally,handle不仅可以获取supplyAsync中的结果,还可以获取异常信息并返回默认结果
handleAsync方法和whenCompleteAsync方法类似,代表其中的代码可能会使用与之前supplyAsync中使用的线程不同的线程执行
使用handle,没有异常的情况:
// 首先创建一个线程池,创建方法参照之前的代码,这里直接封装成方法
ThreadPoolExecutor threadPoolExecutor = createExector();
// 开始异步编排
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
// 异步任务,相当于run()方法
System.out.println("当前线程名:" + Thread.currentThread().getName());
int i = 10 / 2
System.out.println("运行结果:" + i);
return i;
}, threadPoolExecutor).handle((result, exception) -> {
if(exception==null){
System.out.println("没有发生异常,结果是:" + result);
return result;
}else{
System.out.println("发生了异常:" + exception);
return 10;
}
});
// 获取异步任务结果
Integer result = future.get();
System.out.println("main...end..."+result);
使用handle,有异常的情况:
// 制造异常
int i = 10 / 0;
线程串行化
两任务组合
都完成时
两个任务都完成时,才进行下面的操作
runAfterBothAsync():两任务执行完成后再执行,但是不能获取两个任务的执行结果
// 首先创建一个线程池,创建方法参照之前的代码,这里直接封装成方法
ThreadPoolExecutor threadPoolExecutor = createExector();
// 任务1
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务1开始");
int i = 10/2;
System.out.println("任务1结束");
return i;
},threadPoolExecutor);
// 任务2
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务2开始");
int j = 18/3;
System.out.println("任务2结束");
return j;
},threadPoolExecutor);
// 两任务组合
future1.runAfterBothAsync(future2,()->{
System.out.println("两任务组合...");
},threadPoolExecutor);
thenAcceptBothAsync():可以接收到两个任务执行完之后的结果信息
// 首先创建一个线程池,创建方法参照之前的代码,这里直接封装成方法
ThreadPoolExecutor threadPoolExecutor = createExector();
// 任务1
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务1开始");
int i = 10/2;
System.out.println("任务1结束");
return i;
},threadPoolExecutor);
// 任务2
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务2开始");
int j = 18/3;
System.out.println("任务2结束");
return j;
},threadPoolExecutor);
// 两任务组合
future1.thenAcceptBothAsync(future2,(f1,f2)->{
System.out.printf("之前的结果:f1 = %d,f2 = %d\n",f1,f2);
},threadPoolExecutor);
thenCombineAsync():不仅可以获取两个任务的结果,还可以return合并后任务的返回值
// 首先创建一个线程池,创建方法参照之前的代码,这里直接封装成方法
ThreadPoolExecutor threadPoolExecutor = createExector();
// 任务1
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务1开始");
int i = 10/2;
System.out.println("任务1结束");
return i;
},threadPoolExecutor);
// 任务2
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务2开始");
int j = 18/3;
System.out.println("任务2结束");
return j;
},threadPoolExecutor);
// 两任务组合
CompletableFuture<String> future3 = future1.thenCombineAsync(future2, (f1, f2) -> {
return f1 + " -> " + f2;
}, threadPoolExecutor);
System.out.println(future3.get());
一个完成时
两个任务中,只要有一个完成,就可以进行下面的操作
runAfterEitherAsync():无传入参数,无返回值
// 首先创建一个线程池,创建方法参照之前的代码,这里直接封装成方法
ThreadPoolExecutor threadPoolExecutor = createExector();
// 任务1
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务1开始");
int i = 10/2;
System.out.println("任务1结束");
return i;
},threadPoolExecutor);
// 任务2
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务2开始");
int j = 18/3;
try {
// 让任务2延迟3秒
Thread.sleep(3000L);
System.out.println("任务2结束");
} catch (InterruptedException e) {
e.printStackTrace();
}
return j;
},threadPoolExecutor);
// 两任务只要有一个完成,就执行任务3
future1.runAfterEitherAsync(future2,()->{
System.out.println("任务3执行...");
},threadPoolExecutor);
acceptEitherAsync():有传入参数,无返回结果
// 首先创建一个线程池,创建方法参照之前的代码,这里直接封装成方法
ThreadPoolExecutor threadPoolExecutor = createExector();
// 任务1
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务1开始");
int i = 10/2;
System.out.println("任务1结束");
return i;
},threadPoolExecutor);
// 任务2
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务2开始");
int j = 18/3;
try {
// 让任务2延迟3秒
Thread.sleep(3000L);
System.out.println("任务2结束");
} catch (InterruptedException e) {
e.printStackTrace();
}
return j;
},threadPoolExecutor);
// 两任务只要有一个完成,就执行任务3
future1.acceptEitherAsync(future2,(result)->{
System.out.println("任务3执行:"+result);
},threadPoolExecutor);
applyToEitherAsync():有传入参数,有返回值
// 首先创建一个线程池,创建方法参照之前的代码,这里直接封装成方法
ThreadPoolExecutor threadPoolExecutor = createExector();
// 任务1
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务1开始");
int i = 10/2;
System.out.println("任务1结束");
return i;
},threadPoolExecutor);
// 任务2
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务2开始");
int j = 18/3;
try {
// 让任务2延迟3秒
Thread.sleep(3000L);
System.out.println("任务2结束");
} catch (InterruptedException e) {
e.printStackTrace();
}
return j;
},threadPoolExecutor);
// 两任务只要有一个完成,就执行任务3
CompletableFuture<String> future3 = future1.applyToEitherAsync(future2, (result) -> {
return "之前的结果是:" + result;
}, threadPoolExecutor);
System.out.println("任务3的结果是:"+future3.get());
多任务组合
allOf():所有任务执行完,才进行下面的操作
// 首先创建一个线程池,创建方法参照之前的代码,这里直接封装成方法
ThreadPoolExecutor threadPoolExecutor = createExector();
// 任务1
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务1开始");
int i = 10/2;
System.out.println("任务1结束");
return i;
},threadPoolExecutor);
// 任务2
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务2开始");
int i = 21 / 3;
System.out.println("任务2结束");
return i;
}, threadPoolExecutor);
// 任务3
CompletableFuture<Integer> future3 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务3开始");
int j = 18/3;
try {
// 让任务2延迟3秒
Thread.sleep(3000L);
System.out.println("任务3结束");
} catch (InterruptedException e) {
e.printStackTrace();
}
return j;
},threadPoolExecutor);
// 所有任务都执行完成,才进行下面的操作
CompletableFuture<Void> allOf = CompletableFuture.allOf(future1, future2, future3);
// 阻塞等待所有任务都执行完成,这句话必须要加,否则起不到阻塞等待所有任务完成的作用
allOf.get();
// 输出结果
System.out.printf("任务1结果:%d,任务2结果:%d,任务3结果:%d\n",future1.get(),future2.get(),future3.get());
anyOf():只要有一个任务完成,就可以执行下面的操作
// 首先创建一个线程池,创建方法参照之前的代码,这里直接封装成方法
ThreadPoolExecutor threadPoolExecutor = createExector();
// 任务1
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务1开始");
int i = 10/2;
System.out.println("任务1结束");
return i;
},threadPoolExecutor);
// 任务2
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务2开始");
int i = 21 / 3;
System.out.println("任务2结束");
return i;
}, threadPoolExecutor);
// 任务3
CompletableFuture<Integer> future3 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务3开始");
int j = 18/3;
try {
// 让任务2延迟3秒
Thread.sleep(3000L);
System.out.println("任务3结束");
} catch (InterruptedException e) {
e.printStackTrace();
}
return j;
},threadPoolExecutor);
// 只要有一个任务执行完成,就可以进行下面的操作
CompletableFuture<Object> anyOf = CompletableFuture.anyOf(future1, future2, future3);
// 阻塞等待至少有一个任务执行完成
anyOf.get();
// 输出结果
System.out.printf("首先完成的任务的结果是:%d\n",anyOf.get());