如果要讲清楚响应式编程的来龙去脉，最好从最基础的同步编程开始，一步步看到多线程、异步编程，最后再进入响应式编程的世界。这个过程本身就是编程思维不断进化的缩影。
简介
这篇文章会从头梳理这一脉络。从同步编程讲起，到多线程编程的出现，再到为了解决线程阻塞、资源浪费而诞生的异步编程，以及为了让异步编程更好写、更好维护而出现的响应式编程。
1. 示例
我们用同一个实际编程的例子来感受这几种编程方式的区别。
假设要执行一个任务A：先调用功能M算出一个结果，同时调用功能N算出另一个结果，然后把两个结果做某种计算，得到最终结果。M和N都需要一定的时间（比如有阻塞的IO操作）。
接下来我们会用不同的编程模式来实现同样的功能，重点感受每种模式的特点。
2. 同步编程
最简单也最直接的思路：单线程、顺序执行。
代码写出来就是一步一步的：先通过功能M算出m，再通过功能N算出n，然后计算m+n。M和N各自需要1000ms的等待时间。
public static int functionM(){
try {
// calculate m...
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 1;
}
public static int functionN(){
try {
// calculate n...
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 2;
}主线程执行如下：
public static void main(String[] args) throws InterruptedException {
long start = System.currentTimeMillis();
int m = functionM();
int n = functionN();
//Complex calculations
Thread.sleep(1000);
System.out.println("计算结果："+(m+n));
System.out.println("耗时："+(System.currentTimeMillis()-start));
}结果是：耗时3036ms，接近3秒。这就是单线程顺序执行的结果，M和N依次等待，浪费了大量时间。
3. 多线程编程
仔细观察任务A：M和N的执行其实是相互独立的，完全可以同时进行。于是我们想到了多线程并行。
任务A可以拆解成B、C、D三个子任务：B执行M，C执行N，D等待B和C的结果再进行计算。B和C之间没有依赖关系。
实现方式就是给B和C各自分配一个子线程，主线程通过Future等待结果，然后执行D。
public class MultithreadingDemo {
public static void main(String[] args) throws InterruptedException, ExecutionException {
long start = System.currentTimeMillis();
ExecutorService threadpool = Executors.newCachedThreadPool();
Future FutureTaskB = threadpool.submit(MultithreadingDemo::TaskB);
Future FutureTaskC = threadpool.submit(MultithreadingDemo::TaskC);
int m = FutureTaskB.get();
int n = FutureTaskC.get();
int result = TaskD(m, n);
System.out.println("计算结果：" + result);
System.out.println("耗时：" + (System.currentTimeMillis() - start));
threadpool.shutdown();
}
public static int TaskB() { return functionM(); }
public static int TaskC() { return functionN(); }
public static int TaskD(int m, int n) throws InterruptedException {
Thread.sleep(1000);
return m + n;
}
} 运行结果：耗时2078ms，比同步方式快了将近1秒。这就是多线程并行的效果。
4. 异步编程
多线程虽然解决了并行，但仔细看上面的代码：主线程在调用FutureTaskB.get()时是阻塞的——它一直在等子线程的结果，这段时间内什么都干不了。如果业务流量很高，线程是一种宝贵资源，大量线程这样空等着，很容易耗尽系统内存。
于是异步编程的思路出现了：主线程发出任务指令后直接返回，不等结果。等B和C执行完了，再安排其他线程（不一定是原来的线程）去执行后续操作。我们的关注点从“谁执行、什么时候执行”转移到“任务之间的依赖关系”上。
Ja va 8的CompletableFuture是一个很好的工具。我们用它来对这个过程建模：主线程只负责编排任务，然后就可以去做别的事情，不必等着。
任务依赖关系图：
代码示例如下：
public static void main(String[] args) throws InterruptedException, ExecutionException {
triggerA();
//处理其他任务，完全不用管理任务A。
Thread.sleep(3000);
}
public static void triggerA() {
long start = System.currentTimeMillis();
System.out.println("编排任务A线程ID为:"+Thread.currentThread().getId());
CompletableFuture taskBFuture = CompletableFuture.supplyAsync(()->{
System.out.println("执行任务B的当前线程ID:"+Thread.currentThread().getId());
return taskB();
});
CompletableFuture taskCFuture = CompletableFuture.supplyAsync(()->{
System.out.println("执行任务C的当前线程ID:"+Thread.currentThread().getId());
return taskC();
});
//当任务C、D完成之后，回调执行任务D
taskBFuture.thenCombineAsync(taskCFuture,(m,n)->{
try {
System.out.println("执行任务C的当前线程ID:"+Thread.currentThread().getId());
Thread.sleep(1000);
System.out.println("耗时:"+(System.currentTimeMillis()-start));
System.out.println("taskA计算结果:"+(m+n));
} catch (InterruptedException e) {
e.printStackTrace();
}
return null;
});
}
public static int taskB() { return functionM(); }
public static int taskC() { return functionN(); } 运行结果：耗时2042ms，线程1（主线程）编排完任务后就去忙别的了，B和C分别在11、12线程执行，然后由11线程执行D。主线程没有被阻塞。
这里还引出一个更直观的例子：C#的异步编程。三个任务，每个任务分为前、中、后三部分，中间部分是等待操作（如IO）。传统思路下线程会阻塞在等待操作上；异步编程则让线程直接返回，去执行其他任务，等等待完成了再派一个线程去执行后续操作。
Ja va中的非阻塞IO也是类似思想：线程不等待IO操作，而是通过事件机制在IO完成后唤醒执行后续操作。这种模式在高并发场景下效果显著，极大解放了线程资源。
5. 响应式编程
5.1 响应式的由来
从上面的演变可以看到，编码的视角在逐渐从“控制任务执行”转向“描述任务本身和任务之间的依赖关系”。我们越来越关注“事情本身”。
但之前举的例子都很简单。如果业务逻辑复杂，任务的编排也会变得复杂、难以实现。这就是响应式编程要解决的问题。
回过头看我们拆解的任务B、C、D：B和C完成之后回调D执行。与其叫“回调”，不如叫“触发”——B、C的执行结束事件触发了D的执行。换句话说，我们可以看作D订阅了B、C的结束事件。当事件发生时，D开始响应并执行。
发布者发布各种事件（包括BC执行结束的事件），订阅者订阅这些事件。BC执行结束后发布事件通知订阅者，订阅者响应执行D。这就是用订阅-发布模式来表现异步编程。
这么做的好处是：在响应式编程中，一系列事件（产生事件的地方）被看作流或管道。利用函数式编程中对流的处理，可以把异步编程中的许多操作转化为对数据流的操作，尤其是在对流进行转换时，效率和优雅度都会提升一个台阶。
拿Project Reactor举个例子。有一个管道每隔100ms发布一个数字（相当于事件），一个订阅者负责打印这些数字。传统异步操作肯定会阻塞等待，浪费线程资源。改用响应式编程：
Flux.range(1,10)
.delayElements(Duration.of(100,ChronoUnit.MILLIS))
.subscribe(System.out::println);这段代码的意思是：声明一个可以生成1到10数字的管道（Flux.range(1,10)），每隔100ms生成一个数字（delayElements），打印服务订阅这个管道（subscribe），有元素生成时就通知打印服务进行打印（System.out::println）。
为了更直观地对比不同风格的异步编程，再看两个Project Reactor官方的例子。
5.2 回调 vs 响应式
需求：展示五个商品——优先展示用户最喜爱的五个商品，如果没有最喜爱商品，就用推荐的商品展示。
使用回调实现（可用Gua va实现）：
userService.getFa vorites(userId, new Callback>() {
public void onSuccess(List list) {
if (list.isEmpty()) {
suggestionService.getSuggestions(new Callback>() {
public void onSuccess(List list) {
UiUtils.submitOnUiThread(() -> {
list.stream().limit(5).forEach(uiList::show);
});
}
public void onError(Throwable error) {
UiUtils.errorPopup(error);
}
});
} else {
list.stream()
.limit(5)
.forEach(fa vId -> fa voriteService.getDetails(fa vId,
new Callback() {
public void onSuccess(Fa vorite details) {
UiUtils.submitOnUiThread(() -> uiList.show(details));
}
public void onError(Throwable error) {
UiUtils.errorPopup(error);
}
}));
}
}
public void onError(Throwable error) {
UiUtils.errorPopup(error);
}
}); 一层层的回调嵌套，对于开发和维护来说简直是地狱——这被称为“回调地狱”（Callback Hell）。命令式编程可读性极差，层层缩进，难以维护，容易出错。
使用响应式编程实现：
userService.getFa vorites(userId)
.flatMap(fa voriteService::getDetails)
.switchIfEmpty(suggestionService.getSuggestions())
.take(5)
.publishOn(UiUtils.uiThreadScheduler())
.subscribe(uiList::show, UiUtils::errorPopup);流程清晰，转换操作简单，底层异步实现被完全屏蔽。只需要把要做什么表达出来，描述好任务之间的关系，功能就实现了。
5.3 CompletableFuture 响应式
再看一个例子：用一个ID队列，分别获取每个ID的name和statistic，然后组合起来。这是典型的异步编排场景。
CompletableFuture实现：
CompletableFuture> ids = ifhIds();
CompletableFuture> result = ids.thenComposeAsync(l -> {
Stream> zip = l.stream().map(i -> {
CompletableFuture nameTask = ifhName(i);
CompletableFuture statTask = ifhStat(i);
return nameTask.thenCombineAsync(statTask,
(name, stat) -> "Name " + name + " has stats " + stat);
});
List> combinationList = zip.collect(Collectors.toList());
CompletableFuture[] combinationArray =
combinationList.toArray(new CompletableFuture[combinationList.size()]);
CompletableFuture allDone = CompletableFuture.allOf(combinationArray);
return allDone.thenApply(v ->
combinationList.stream().map(CompletableFuture::join)
.collect(Collectors.toList()));
});
List results = result.join();
assertThat(results).contains(...); 虽然比原生回调简化了一些，但读起来依然比较吃力。任务依赖关系图可以帮助理解：
响应式编程实现：
Flux ids = ifhrIds();
Flux combinations = ids.flatMap(id -> {
Mono nameTask = ifhrName(id);
Mono statTask = ifhrStat(id);
return nameTask.zipWith(statTask,
(name, stat) -> "Name " + name + " has stats " + stat);
});
Mono> result = combinations.collectList();
List results = result.block();
assertThat(results).containsExactly(...); FLUX是可以生成0到N个元素的管道，Mono是可以生成0到1个元素的管道。代码的声明性更强，对任务编排的API也更友好，开发和理解的难度明显降低。
从回调 → CompletableFuture → 响应式编程，代码的可维护性、声明性和面对复杂业务的灵活性都在逐步提高。
最后用响应式编程改造一下一开始那个异步编程的例子：
public class ReactiveProgramDemo {
public static void main(String[] args) throws InterruptedException {
long start = System.currentTimeMillis();
Mono monoB = Mono.fromCallable(ReactiveProgramDemo::taskB)
.publishOn(Schedulers.boundedElastic());
Mono monoC = Mono.fromCallable(ReactiveProgramDemo::taskC)
.publishOn(Schedulers.boundedElastic());
Mono.zip(monoB, monoC)
.subscribeOn(Schedulers.boundedElastic())
.subscribe(nums -> {
System.out.println("计算结果为：" + taskD(nums.getT1(), nums.getT2()));
System.out.println("tasA执行用时：" + (System.currentTimeMillis() - start));
});
//主线程做其他的事情
Thread.sleep(5000);
}
// ... taskB, taskC, taskD 同上
} 结果：耗时2249ms，一切异步执行，主线程不被阻塞。
5.4 响应式编程的特性
随着响应式编程的普及，reactive-streams规范也逐渐成形，现在的响应式框架都是按照这个规范进行接口实现的。概括一下它的核心特性：
- 异步编程的实现：响应式编程本质上就是一种异步编程，与基于回调、CompletableFuture实现的异步编程属于同一范畴。
- 更好的表达性，不易出错：用声明式的方式描述业务逻辑，而不是控制流程，自然减少了出错概率。
- 更灵活的操作方式：得益于数据流的函数式编程，提供了丰富的转换和操作符。
- 更好的异常处理：其他方式的异步编程很难做到优雅的异常处理，响应式编程可以轻松实现。
Flux.range(1,100)
.map(x->1/(x%10))
.doOnNext(System.out::println)
.doOnError(System.err::println)
.subscribe()- 支持背压：可以控制上游管道的发送速度，避免下游消费不过来。
这些特性仅仅是入门，后续的文章会针对响应式编程进行更详细的讲解。
6. 总结
从最初的问题出发，我们经历了几个阶段：
先是同步编程——顺序执行，简单直接，但不能避免等待。
然后发现有些地方可以并行执行，于是出现了多线程编程，但依然没有摆脱“同步等待”的低效率问题。
随着非阻塞IO和异步接口（如Ja va的NIO和CompletableFuture）的普及，异步编程流行起来。它大大提高了线程的利用率，让服务在更少的时间内完成更多计算，性能大幅提升。
但新的问题随之而来：异步编程在面对复杂业务时的设计难度大，任务之间的关系复杂。用传统方式实现异步编程，容易出现冗长的嵌套代码，维护成本高，而且对复杂业务的任务编排十分困难。
这时候，封装更完善的响应式编程出现了。它从事件的角度出发，将异步编程转化为对事件流的操作，利用函数式编程对数据流进行处理，让异步编程变得更加优雅。同时提供了灵活的转换操作、异常处理和背压机制。对于异步编程来说，这确实是令人兴奋的进步。
7. 思考
语言或框架的接口提供了异步的能力，它们到底是怎么实现的？
8. 参考
一文带你彻底了解ja va异步编程
ja va-8-completablefuture-tutorial
Reactor 3 Reference Guide
Difference Between Asynchronous Programming and Multithreading in C#
What is the difference between asynchronous programming and multithreading?
