异步编程在对响应时间近乎严苛的今天，受到了越来越多的关注，尤其是在IO密集型业务中。对比传统的同步模式，异步编程可以提高服务器的响应时间和处理业务的能力，从而达到快速给用户响应的效果。
代码前置：方法中会直接使用到线程池和print函数
public class TestAsync {
// 创建一个线程池，大小为10
ExecutorService executorService = Executors.newFixedThreadPool(10);
private String printThread(String message) {
return Thread.currentThread().getName() + " " + message;
}
//下面的示例代码...
}
- java.util.concurrent.Future是JDK5引入的，用来获取一个异步计算的结果。可以使用isDone方法检查计算是否完成，也可以使用get阻塞住调用线程，直到计算完成返回结果，使用cancel方法停止任务的执行。
- FutureTask.java是对Futre和Runnable最简单的实现，实现了run函数，所以可以直接执行，任务执行结束通过set()保存结果，setException()保存异常信息。通常配合executorService.submit()一起使用，ExecutorService中将任务包装成FutureTask执行execute()；
样例如下图：
public class TestAsync {
// 创建一个线程池，大小为10
ExecutorService executorService = Executors.newFixedThreadPool(10);
private String printThread(String message) {
return Thread.currentThread().getName() + " " + message;
}
/**
* 小明需要等待厨师炒好菜
* @throws InterruptedException
* @throws ExecutionException
*/
@Test
public void futureCallBackTest() throws InterruptedException, ExecutionException {
long start = System.currentTimeMillis();
System.out.println(printThread("小明点餐"));
Future<String> future = executorService.submit(() -> {
System.out.println(printThread("厨师开始炒菜"));
Thread.sleep(2000);
System.out.println(printThread( "厨师炒好菜"));
return "饭菜好了";
});
//休眠5s，模拟其他业务代码执行【可以执行其他操作，而不受厨师炒菜的影响】
Thread.sleep(5000);
String result = future.get();
executorService.shutdown();
System.out.println(printThread(result + ",小明开始吃饭"));
//long end = System.currentTimeMillis();
//long res = end - start;
//System.out.println("耗费时间：" + res);
}
}
- 能获取异步线程执行结果
- 无法方便得知任务何时完成
- 在主线程获取任务的过程中会导致主线程阻塞
- 复杂一点的情况下，比如多个异步任务的场景，一个异步任务依赖上一个异步任务的执行结果，异步任务合并等，Future无法满足需求
- Google框架中的Guava 提供了 ListenableFuture 类来执行异步操作
- ListenableFuture对Java原生的Future做了扩展，顾名思义就是使用了监听器模式实现了回调
- 通过addListener(Runnable listener, Executor executor)方法添加回调任务。
- 要使用listenableFuture还要结合MoreExecutor线程池，MoreExecutor是对Java原生线程池的封装，比如常用的MoreExecutors.listeningDecorator(threadPool); 修改Java原生线程池的submit方法，封装了future返回listenableFuture。
添加maven依赖：
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>28.2-jre</version>
</dependency>
样例：
/**
* Google的guava
* @throws InterruptedException
* @throws ExecutionException
*/
@Test
public void listenableFutureTest() throws InterruptedException, ExecutionException {
System.out.println(printThread("小明点餐"));
ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
ListenableFuture<String> listenableFuture = listeningExecutorService.submit(() -> {
System.out.println(printThread("厨师开始炒菜"));
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(printThread( "厨师炒好菜"));
return "饭菜好了";
});
Futures.addCallback(listenableFuture, new FutureCallback<String>() {
@Override
public void onSuccess(@Nullable String s) {
System.out.println(printThread(s + ",小明开始吃饭"));
}
@Override
public void onFailure(Throwable throwable) {
System.out.println(printThread( throwable.getMessage()));
}
}, executorService);
System.out.println(printThread( "小明开始玩游戏"));
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(printThread("小明结束玩游戏"));
listenableFuture.get();
listeningExecutorService.shutdown();
executorService.shutdown();
}
这里除main线程外，重新创建了两个线程，没有阻塞等待
- 充分利用了时间片，只要任务结束就马上响应
- 但是可能会导致Callback Hell（回调地狱）
例如，我们可以通过回调实现如下逻辑：
但是该方法会导致代码难以理解同时不易于维护，由于代码过多，此处以图展现
- Java8的新类，借鉴了Google Guava的ListenableFuture
- CompleteableFuture可以用声明式语义来创建多种异步任务：
- 创建一个异步任务
- 创建一个异步任务，并且该任务必须在一个前驱异步任务之后执行，其以前驱任务的输出作为自身的输入
- 创建一个异步任务，且该任务必须在若干个前驱异步任务中的（任意或全部）完成之后执行，其以全部（或任一）前驱任务的输出作为自身的输入
样例：
@Test
public void completeableFutureTest() {
System.out.println(printThread("小明点餐"));
CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
System.out.println(printThread("厨师开始做菜"));
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(printThread("厨师菜做好了"));
return "菜已装盘";
});
CompletableFuture<Void> completableFuture1 = CompletableFuture.runAsync(() -> {
System.out.println(printThread( "小明开始玩游戏"));
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(printThread("小明结束玩游戏"));
});
CompletableFuture<Void> completableFuture2 = completableFuture
.thenAcceptBoth(completableFuture1,(a, b) -> System.out.println(printThread( a + ", 小明开始吃饭,并点了饮料")))
.thenApplyAsync((b) -> {
System.out.println(printThread("服务员拿饮料"));
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "饮料好了";
},executorService)
.thenAcceptAsync((s) -> System.out.println(printThread(s + ",小明开始喝饮料")));
completableFuture2.join();
}
可以看到我们可以让通过监听的方式让线程实现某种业务顺序
- Reactor 框架是 Pivotal 公司（ Spring 家族公司）开发的，实现了 Reactive Programming 思想，符合 Reactive Streams 规范的一项技术；
- Reactor 是一个高度可伸缩、内存利用率高、灵活性强的响应式编程框架，有助于提高应用程序的性能和稳定性。
- Reactor 是基于流式编程的一种响应式编程框架。它实现了响应式流 API，可以处理并发数据的异步流，并支持通过多种操作符进行流式转换和操作。与传统的命令式编程范式不同，响应式编程可以更容易地实现异步、响应式、非阻塞和可响应式等特性，并且在处理IO密集型应用程序时具有明显的优势。
- 主要接口：
- Publisher
- Subscriber
- Subcription
其中，Subcriber 中便包含了非常重要的 onNext、onError、onCompleted 这三个方法。
导入maven依赖：
<!--reactor-->
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
<version>3.2.3.RELEASE</version>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<version>3.2.3.RELEASE</version>
<scope>test</scope>
</dependency>
public class TestReactor {
public static void main(String[] args) throws InterruptedException {
Flux.just(1, 2, 3, 4, 5)
.subscribeOn(Schedulers.parallel())
.subscribe(new CoreSubscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
System.out.println(printThread("onSubscribe, " + s.getClass().toString()));
s.request(5);
}
@Override
public void onNext(Integer integer) {
System.out.println(printThread("next: " + integer));
}
@Override
public void onError(Throwable t) {
}
@Override
public void onComplete() {
System.out.println(printThread("complete"));
}
});
Thread.sleep(1000);
}
private static String printThread(String note) {
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("hh:mm:ss");
long time = System.currentTimeMillis();
Date date = new Date(time);
return Thread.currentThread().getName() + " " + simpleDateFormat.format(date) + " " + time + " " + note;
}
}
可以看到输出结果是按照一定顺序的
- 我们可以根据流式编程来处理很多复杂的业务逻辑
回调函数机制：回调函数是一种面向事件或消息的程序设计方法，它通过将函数对象作为参数传递给其他函数或对象来实现异步操作。在 Java 中，可以使用接口或匿名类来实现回调函数机制。
public class AsyncTask {
// 定义回调接口
public interface Callback {
void onCompleted(String result);
}
// 模拟一个耗时操作
public void doTask(final Callback callback) {
new Thread(new Runnable() {
@Override
public void run() {
// 在新的线程中执行异步任务，模拟一个耗时操作，即等待 3 秒
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 返回一个结果给回调函数
callback.onCompleted("AsyncTask completed!");
}
}).start();
}
// 测试代码
public static void main(String[] args) {
// 创建一个 AsyncTask 实例
AsyncTask asyncTask = new AsyncTask();
// 调用 doTask 方法并传入回调函数
asyncTask.doTask(new Callback() {
@Override
public void onCompleted(String result) {
// 在回调函数中打印异步任务的结果
System.out.println(result);
}
});
// 再做一些其他操作
// ...
}
}
- AsyncTask 类中的 doTask 方法模拟了一个耗时操作，并通过回调函数的方式实现了异步。在主程序中，我们创建一个 AsyncTask 实例并调用它的 doTask 方法来执行异步任务，同时传入一个回调函数，在回调函数中打印异步任务的结果。在执行异步任务的同时，我们还可以继续进行其他操作。这样可以避免阻塞主线程，提高程序的并发性能。
- 总之，回调函数是 Java 中一种实现异步操作的简单、方便的方式，可以帮助我们在异步执行的任务完成后及时处理结果，提高程序的响应性和并发性能。
回调函数机制：回调函数是一种面向事件或消息的程序设计方法，它通过将函数对象作为参数传递给其他函数或对象来实现异步操作。在 Java 中，可以使用接口或匿名类来实现回调函数机制。
Future 和 CompletableFuture：Future 是 Java 中的一种异步编程模型，它允许调用者异步地等待另一个线程的执行结果。CompletableFuture 是 Java 8 引入的一种强化版 Future，可以更方便地实现异步事件的组合、异常处理等操作。
线程池和多线程：在 Java 中，可以使用线程池和多线程来异步执行任务。通过将任务提交给线程池，可以让执行过程异步地执行，并且可以利用多个线程来处理任务，提高程序的并发性能。
RxJava：RxJava 是一个基于 ReactiveX 编程模型的响应式编程库，在处理异步事件流时具有很高的效率和灵活性。RxJava 提供了多种操作符用于处理数据流，包括过滤、转换、组合等。
参考文章：
https://blog.csdn.net/mowushenght/article/details/122211330