响应式编程详解(看这篇就够了)

响应式编程详解(看这篇就够了)-mikechen

为了应对高并发开发场景，一个更优雅地实现异步编程的方式 Reactive Programming出现了 ，我们称之为响应式编程。

什么是响应式编程

响应式编程是一款使用异步数据流编程的响应式编程思想，是基于观察者模型的这是大家的共识，它提供了非阻塞、异步的特性，便于处理异步情景，从而避免回调地狱和突破Future的局限性。

响应式编程 可以理解为：当某一主题发生改变时，观察此主题的观察者就会立刻收到通知并做出一系列响应。

可见，观察者模式的概念比较广，是“相应改变”，即只要变化了就行；而响应式编程必须得是“一系列响应”，也就是不但要变化，还要像雪崩一样的触发连锁式变化。

举个响应式编程的例子：在一个“数据监控系统”中，如果“数据”发生了改变，就会触发一系列的变化：数据改变 -> Dao层做出响应（数据访问层）-> Service层做出响应（业务逻辑层）-> Controller做出响应（控制器） -> Web页面做出相应（UI）。

至此，应该能理解响应式编程的概念了。

 

为什么要使用响应式编程

在JVM上提供了两种异步编程模型：

Callbacks：异步方法没有返回值，但是需要一个额外的回调参数（lambda或者匿名函数），当结果可用时调用这个参数，一个著名的例子是Swing’s的EventListener层次结构。

Futures：异步方法立即返回一个Future。

异步进程计算T值，通过Future对象包装对T值的访问。该值不是立即可用的，可以轮询该对象，直到该值可用为止。例如，ExecutorService使用Future对象，运行Callable<T任务。

但是这两种技术都有他们的局限性，回调难以组合在一起，很快就会导致代码难以阅读和维护（这种情况称为回调地狱（Callback Hell））。

尽管Java 8通过CompletableFuture进行了改进，编排多个Future对象是可行的，但是这并不容易，并且Future还有其他问题：

  • 调用get()方法很容易导致Future对象出现另一种阻塞情况。
  • 不支持惰性计算。
  • 缺乏对多值和高级错误处理的支持。

响应式库，如Reactor，Rxjava旨在解决JVM上”经典”异步方法的这些缺点。

 

响应式编程的优点

响应式开发的好处主要包含以下几点：

  • 提高所开发程序的性能。
  • 在多核机器上，提高了计算资源的利用率。
  • 为异步编程提供了一个更靠谱的可维护方案。
  • 提供了背压机制，也就是对计算资源提供了过载保护功能。

 

Java响应式编程

响应式编程会用到一个发布者和一个订阅者，然后通过订阅关系完成数据流的传输。订阅关系中可以处理一些背压问题，即调节消费者与生产者之间的供需平衡，让整个程序达到最大效率。

响应式编程详解(看这篇就够了)-mikechen

Java9中java.util.concurrent.Flow接口提供响应式流编程类似的功能。

下面我们实现一个基于Java 响应式编程的示例：

其中有三个简单步骤：

  1. 建立生产者
  2. 构建消费者
  3. 消费者订阅生产者
  4. 生产者生产内容
SubmissionPublisher publisher = new SubmissionPublisher<>();//建立生产者
Flow.Subscriber subscriber = new Flow.Subscriber() {...};//建立消费者 (其中的实现放到下面)
publisher.subscribe(subscriber);//订阅关系
for (int i = 0; i < 10; i++) {
 publisher.submit("test reactive java : " +i); //生产者生产内容
}

消费者全部代码如下：

Flow.Subscriber subscriber = new Flow.Subscriber() {
    Flow.Subscription subscription;
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("Subscription establish first ");
        this.subscription = subscription;
        this.subscription.request(1);
    }
    @Override
    public void onNext(Object item) {
        subscription.request(10);
        System.out.println("receive :  "+ item);
    }
    @Override
    public void onError(Throwable throwable) {
        System.out.println(" onError ");
    }
    @Override
    public void onComplete() {
        System.out.println(" onComplete ");
    }
};

其中onSubscribe方法表示建立订阅关系

onNext接受数据，并请求生产者的数据。

onError，onComplete则是error或者完成之后的处理方法。

带有中间处理器的响应式流

Reactive Stream 通常会基于如下的模型：

响应式编程详解(看这篇就够了)-mikechen

下面我们实现一个带有中间处理功能的响应式模型：

下面的Processor 既有发布者，又有订阅者：

public class ReactiveProcessor extends SubmissionPublisher implements Flow.Subscriber {
    private Flow.Subscription subscription;
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println( Thread.currentThread().getName() +  " Reactive processor establish connection ");
        this.subscription = subscription;
        this.subscription.request(1);
    }

    @Override
    public void onNext(Object item) {
        System.out.println(Thread.currentThread().getName() + " Reactive processor receive data: "+ item);
        this.submit(item.toString().toUpperCase());
        this.subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("Reactive processor error ");
        throwable.printStackTrace();
        this.subscription.cancel();
    }

    @Override
    public void onComplete() {
        System.out.println(Thread.currentThread().getName() + " Reactive processor receive data complete ");
    }
}

如上中间处理器订阅发布者， 同时消费者再订阅中间处理器。中间处理器也可以调节发布订阅的生产消费速率。

SubmissionPublisher publisher = new SubmissionPublisher<>(); //创建生产者
ReactiveProcessor reactiveProcessor = new ReactiveProcessor(); // 创建中间处理器
publisher.subscribe(reactiveProcessor); //中间处理器订阅生产者
Flow.Subscriber subscriber = new Flow.Subscriber() {...}; //创建消费者
reactiveProcessor.subscribe(subscriber); //消费者订阅中间处理器
for (int i = 0; i < 10; i++) {
    publisher.submit("test reactive java : " +i); //生产者生产数据
}

通过上述生产者-> 中间处理器->消费者， 可以将生产者生产的数据全部变成大写，然后再发送给最终的消费者。

以上式Java中的reactive 编程示例。Java会不同线程来分别处理消费者与生产者的消息处理

Reactor

Reactor中两个比较关键的对象式Flux和Mono， 整个Spring的响应式编程均式基于projectreactor项目。Reactor是响应式编程的依赖，主要是基于JVM构建非阻塞程序。

根据Reactor的介绍，此类响应式编程的的三方库（Reactor）主要是解决一些JVM经典异步编程中的一些缺点，并且还可以专注于一些新的特性，如下：

  • 可组合性与可读性 （Composability and readability）
  • 可以使用丰富的运算操作符将数据作为流进行操作
  • 订阅之前，不会有任何事
  • 背压特性（Backpressure ），可以理解为消费者可以向生产者发送产出率过高的信号，从而调整生产速率。或者消费者可以选择一次性拉去一捆数据进行消费。
  • 于并发无关的高度抽象的高级功能

其中有这么一段解释，可以形象的说明响应式编程。

Reactive的程序可以想象成车间的流水线，reactor既是流水线上的传送带，又是处理工作站。原料从一个原始的生产者出发，最终成为产品被推总给消费者。

 Flux & Mono

下面我们介绍一下Flux和Mono。

在Reactor中Flux和Mono均是Publisher，即生产者。两者也有不同。Flux对象表示0到N个异步的响应序列，而Mono只代表0个（empty）或者1个结果。

Reactor官网上介绍的Flux示意如下：

响应式编程详解(看这篇就够了)-mikechen

Mono示意如下：

响应式编程详解(看这篇就够了)-mikechen

Flux Mono创建与使用

我们也可以单独引用其依赖。

使用maven依赖

<dependencies>
    <dependency>
        <groupId>io.projectreactor</groupId>
        <artifactId>reactor-core</artifactId> 
    </dependency>
    <dependency>
        <groupId>io.projectreactor</groupId>
        <artifactId>reactor-test</artifactId> 
        <scope>test</scope>
    </dependency>
</dependencies>

Mono创建

分别创建空Mono和一个包含一个String的Mono，并由消费者消费打印。

Mono.empty().subscribe(System.out::println);
Mono.just("Hello Mono Java North").subscribe(System.out::print);

Flux创建

Flux创建有如下的一些方法，

  • just（通过不定参数创建）
  • range（从某个整数开始，往后的整数数量）
  • fromArray，fromIterable，fromStream，从名称上就可以看出来，通过数组，迭代器，Stream流创建Flux

下面式一些Java代码示例

Flux.just(1,2,3,4,5).subscribe(System.out::print);
Flux.range(1,20).subscribe(System.out::print);
Flux.fromArray(new String[]{"a1","a2","a3","a4","a5","a6"}).skip(2).subscribe(System.out::print);
Flux.fromIterable(Arrays.asList(1,2,3,4,5,6,7)).subscribe(System.out::println);
Flux.fromStream(Stream.of(Arrays.asList(1,2,3,4,5,6,7))).subscribe(System.out::print);

我们再举一个generate的例子

public static <T, S> Flux<T> generate(Callable<S> stateSupplier, BiFunction<S, SynchronousSink<T>, S> generator)

如上代码所示，generate需要一个Callable参数，而且是supplier （即没有输入值，只有一个输出）

另一个参数是BiFunction （前面我们也介绍过，需要两个输入值，一个输出值）。BiFunction中的其中一个输入值是SynchronousSink，下面我们给出一个generate创建Flux的示例。

Flux.generate(
 () -> 0, //提供一个初始状态值0
 (i, sink) -> {
    sink.next("3*" + i + "=" + 3 * i);//使用初始值去生产一个3的乘法
    if (i > 9) sink.complete();//设置停止条件
    return i + 1;//返回一个新的状态值，以便在下一次的生产中使用，除非响应序列终止
}).subscribe(System.out::println);

下面我们在看一个Flux嵌套处理示例：

需求：将字符串去空格，并去重，然后排序输出。

String str = "qa ws ed rf tg yh uj i k ol p za sx dc vf bg hn jm k loi yt ";
Flux.fromArray(str.split(" "))//通过数组创建Flux
    .flatMap(i -> Flux.fromArray(i.split(""))) 
    .distinct() // 去重
    .sort() //排序
    .subscribe(System.out::print); 
    //flatMap与Stream中的flatMap类似，接受Function作为参数，输入一个值，输出一个值，此处输出均为Publisher，

以上就是Flux和Mono的一些简单介绍，同时Ractor也支持JDK中的FlowPubliser 和FlowSubscriber与 Reactor中的publisher, subscriber的适配等.

WebFlux

SpringBoot 2之后支持的Reactive响应式编程。

关于Reactive技术栈和经典的Servlet技术栈对比，Spring官网的这张图比较清晰。

响应式编程详解(看这篇就够了)-mikechen

Spring响应式编程主要依赖于Reactor第三方库，即上面讲的Flux和Mono的库。

WebFlux主要有以下几个要点：

  • 反应式栈web框架
  • 完全异步非阻塞
  • 运行在netty，undertow，Servlet3.1 + 容器
  • 核心反应式库 Reactor
  • 返回 Flux 或Mono
  • 支持注解和函数编程两种编程模式

Spring WebFlux示例

下面我们给出几个SpringBoot 的响应式web示例。

可以去https://start.spring.io/ 新建webflux的项目也可以。

项目中的主要依赖就是spring-boot-starter-webflux

  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-webflux</artifactId>
  </dependency>

基于注解的WebFlux：

以下是一个最简单的基于注解的WebFlux

@GetMapping("/hello/mono1")
public Mono<String> mono(){
    return Mono.just("Hello Mono -  Java North");
}

@GetMapping("/hello/flux1")
public Flux<String> flux(){
    return Flux.just("Hello Flux","Hello Java North");
}

基于函数式编程的WebFlux:

创建RouterFunction，将其注入到Spring中即可。

@Bean
public RouterFunction<ServerResponse> testRoutes1() {
    return RouterFunctions.route().GET("/flux/function", new HandlerFunction<ServerResponse>() {
        @Override
        public Mono<ServerResponse> handle(ServerRequest request) {
            return ServerResponse.ok().bodyValue("hello web flux , Hello Java North");
        }
    }).build();
}

//上面的方法使用函数式编程替换之后如下
@Bean
public RouterFunction<ServerResponse> testRoutes() {
    return RouterFunctions.route().GET("/flux/function",
         request -> ServerResponse.ok()
                    .bodyValue("Hello web flux , Hello Java North")).build();
}

Flux与Mono的响应式编程延迟示例

下面我们编写一段返回Mono的响应式Web服务。

@GetMapping("/hello/mono")
public Mono<String> stringMono(){
    Mono<String> from = Mono.fromSupplier(() -> {
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "Hello, Spring Reactive  date time:"+ LocalDateTime.now();
    });
    System.out.println( "thread : " + Thread.currentThread().getName()+ " ===  " + LocalDateTime.now() +"  ==========Mono function complete==========");
    return from;
}

使用postman请求如下， 5秒钟后返回数据。后台却在5秒中之前已经处理完整个方法。

响应式编程详解(看这篇就够了)-mikechen

后台打印日志：

响应式编程详解(看这篇就够了)-mikechen

再看一组Flux

@GetMapping(value = "/hello/flux", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> flux1(){
    Flux<String> stringFlux = Flux.fromStream(IntStream.range(1,6).mapToObj(i ->{
        mySleep(1);//表示睡1秒
        return "java north flux" + i + "date time: " +LocalDateTime.now();
    }));
    System.out.println("thread : " + Thread.currentThread().getName()+ " ===  " + LocalDateTime.now() + "  ==========Flux function complete=========");
    return stringFlux;
}

此次使用谷歌浏览器请求此服务：

可以发现每隔一秒就会有一条消息被生产出来。

响应式编程详解(看这篇就够了)-mikechen

后台完成时间同样是在一开始就完成整个方法：

响应式编程详解(看这篇就够了)-mikechen

通过上述对Flux 与 Mono的例子，可以好好体会一下响应式编程。

响应式编程总结

本篇回顾了函数式编程，Stream操作等，然后再举例讲了Java中的Reactive编程示例， 同时也给处理Reactor三方库的Flux于Mono的示例。

最后使用了SpringBoot WebFlux 创建简单的响应式web服务，希望能让大家更好的理解响应式编程。

陈睿mikechen

10年+大厂架构经验，资深技术专家，就职于阿里巴巴、淘宝、百度等一线互联网大厂。

关注「mikechen」公众号，获取更多技术干货！

后台回复【面试】即可获取《史上最全阿里Java面试题总结》，后台回复【架构】，即可获取《阿里架构师进阶专题全部合集》

相关推荐

Java架构师最新学习路线图

阿里P8架构师面试系列专题

手把手带你大型项目实战落地

Java进阶架构师18大技能专题

相关面试

2024最新大厂Spring考题解析

2024最新大厂Java面试题解析

2024大厂MySQL面试题解析

2024大厂Redis面试考题解析

评论交流
    说说你的看法
❯
返回顶部