响应式编程系列（一）：什么是响应式编程？react…

响应式编程 系列文章目录

（一）什么是响应式编程？reactor入门

（二）Flux入门学习：流的概念，特性和基本操作

（三）Flux深入学习：流的高级特性和进阶用法

（四）reactor-core响应式api如何测试和调试？

（五）Spring reactive: Spring WebFlux的使用

（六）Spring reactive: webClient的使用

引言

　　Spring framework 5 的一大新特性：响应式编程（Reactive Programming）。那么什么是响应式？他能给我们带来什么？如何优雅地使用？本系列会从最基础的概念和简单的api讲起，再慢慢深入探讨响应式的一些高级特性，最后讲解实战内容，例如WebFlux和WebClient等在Spring boot中的使用，如何测试和调试。

　　想要了解原理的话，美团点评的这篇博客 Java NIO浅析 非常适合入门。

简单地说：

　　当我们调用socket.read()、socket.write()这类阻塞函数的时候，这类函数不能立即返回，也无法中断，需要等待socket可读或者可写，才会返回，因此一个线程只能处理一个请求。在这等待的过程中，cpu并不干活，（即阻塞住了），那么cpu的资源就没有很好地利用起来。因此对于这种情况，我们使用多线程来提高cpu资源的利用率：在等待的这段时间，就可以切换到别的线程去处理事件，直到socket可读或可写了，通过中断信号通知cpu，再切换回来继续处理数据。例如线程A正在等待socket可读，而线程B已经就绪了，那么就可以先切换到线程B去处理。虽然上下文切换也会花一些时间，但是远比阻塞在线程A这里空等要好。当然计算机内部实际的情况比这复杂得多。

　　而NIO的读写函数可以立刻返回，这就给了我们不开线程利用CPU的最好机会：如果一个连接不能读写（socket.read()返回0或者socket.write()返回0），我们可以把这件事记下来。因此只需要一个线程不断地轮询这些事件，一旦有就绪的时间，处理即可。不需要多线程。

阻塞型IO

• 需要多线程，即需要很大的线程池。
• 每个请求都要有一个单独的线程去处理。

非阻塞型IO

• 只需要数量非常少的线程。
• 固定的几个工作线程去处理事件。

使用NIO我们能得到什么？

• 事件驱动模型
• 避免多线程
• 单线程处理多任务
• 非阻塞I/O，I/O读写不再阻塞，而是返回0
• 基于block的传输，通常比基于流的传输更高效
• 更高级的IO函数，zero-copy
• IO多路复用大大提高了Java网络应用的可伸缩性和实用性

响应式编程入门

　　响应式编程就是基于reactor的思想，当你做一个带有一定延迟的才能够返回的io操作时，不会阻塞，而是立刻返回一个流，并且订阅这个流，当这个流上产生了返回数据，可以立刻得到通知并调用回调函数处理数据。

基本模型

我们首先需要理解响应式编程的基本模型：

Flux

　　Reactor中的发布者（Publisher）由Flux和Mono两个类定义，它们都提供了丰富的操作符（operator）。一个Flux对象代表一个包含0..N个元素的响应式序列，元素可以是普通对象、数据库查询的结果、http响应体，甚至是异常。而一个Mono对象代表一个包含零/一个（0..1）元素的结果。上图就是一个Flux类型的数据流，Flux往流上发送了3个元素，Subscriber通过订阅这个流来接收通知。

如何创建一个流？最简单的方式有以下几种：

//创建一个流，并直接往流上发布一个值为value数据
Flux.just(value);

//通过list创建一个流，往流上依次发布list中的数据
Flux.fromIterable(list);

//创建一个流，并向流上从i开始连续发布n个数据，数据类型为Integer
Flux.range(i, n);

//创建一个流，并定时向流上发布一个数据，数据从0开始递增，数据类型为Long
Flux.interval(Duration.ofSeconds(n));

既然是“数据流”的发布者，Flux和Mono都可以发出三种“数据信号”：元素值、错误信号、完成信号，错误信号和完成信号都是终止信号，完成信号用于告知下游订阅者该数据流正常结束，错误信号终止数据流的同时将错误传递给下游订阅者。

Subscriber 

subscriber是一个订阅者，他只有非常简单的4个接口：

public interface Subscriber<T> {
    void onSubscribe(Subscription var1);

    //收到下一个元素值信号时的行为
    void onNext(T var1);

    //收到错误信号时的行为
    void onError(Throwable var1);

    //收到终止信号时的行为
    void onComplete();
}

Subscriber必须要订阅一个Flux才能够接收通知：

flux.subscribe(
    value -> handleData(value),
    error -> handleError(error),
    () -> handleComplete()
);

上面这个例子通过lambda表达式，定义了Subscriber分别在收到消息，收到错误，和消息流结束时的行为，当Subscriber接收到一个新数据，就会异步地执行handleData方法处理数据。

简单例子：

接下来我们创建几个最简单的流来试一下：

首先我们新建一个maven项目，引入reactor的类库：

<dependencies>
    <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>
</dependencies>

编写代码如下：

public class ReactorTests {

    @After
    public void after() {
        sleep(30_000);
    }

    @Test
    public void testJust() {
        Flux.just("hello", "world")
            .subscribe(System.out::println);
    }

    @Test
    public void testList() {
        List<String> words = Arrays.asList(
            "hello",
            "reactive",
            "world"
        );

        Flux.fromIterable(words)
            .subscribe(System.out::println);
    }

    @Test
    public void testRange() {
        Flux.range(1, 10)
            .subscribe(System.out::println);
    }

    @Test
    public void testInterval() {
        Flux.interval(Duration.ofSeconds(1))
            .subscribe(System.out::println);
    }
}

订阅这些流，收到数据之后只是简单地把它打印出来，运行这些Test，就能够看到订阅者在接收到流上的数据时，异步地去处理这些数据。

标签：

版权申明：本站文章部分自网络，如有侵权，请联系：west999com@outlook.com
特别注意：本站所有转载文章言论不代表本站观点，本站所提供的摄影照片，插画，设计作品，如需使用，请与原作者联系，版权归原作者所有