WebFlux 响应式编程和实现

WebFlux 响应式编程介绍及简单实现

1、响应式编程概述

背景知识

为了应对高并发服务器端开发场景，在2009 年，微软提出了一个更优雅地实现异步编程的方式——Reactive Programming，我们称之为响应式编程。随后，Netflix 和LightBend 公司提供了RxJava 和Akka Stream 等技术，使得Java 平台也有了能够实现响应式编程的框架。

在2017 年9 月28 日，Spring 5 正式发布。Spring 5 发布最大的意义在于，它将响应式编程技术的普及向前推进了一大步。而同时，作为在背后支持Spring 5 响应式编程的框架Spring Reactor，也进入了里程碑式的3.1.0 版本。

什么是响应式编程

响应式编程是一种面向数据流和变化传播的编程范式。这意味着可以在编程语言中很方便地表达静态或动态的数据流，而相关的计算模型会自动将变化的值通过数据流进行传播。

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

电子表格程序就是响应式编程的一个例子。单元格可以包含字面值或类似"=B1 C1"的公式，而包含公式的单元格的值会依据其他单元格的值的变化而变化。

响应式传播核心特点之一：变化传播：一个单元格变化之后，会像多米诺骨牌一样，导致直接和间接引用它的其他单元格均发生相应变化。

2、基于Java8及之前版本的实现

Observable类：此类表示可观察对象，或模型视图范例中的“数据”。 它可以被子类实现以表示应用程序想要观察的对象。

//想要观察的对象 ObserverDemo
public class ObserverDemo extends Observable {
    public static void main(String[] args) {
        ObserverDemo observerDemo = new ObserverDemo();
        //添加观察者
        observerDemo.addObserver((o,arg)->{
            System.out.println("数据发生变化A");
        });
        observerDemo.addObserver((o,arg)->{
            System.out.println("数据发生变化B");
        });
        observerDemo.setChanged();//将此Observable对象标记为已更改
        observerDemo.notifyObservers();//如果该对象发生了变化，则通知其所有观察者
    }
}

启动程序测试：

3、基于 Reactor 实现

Reactor 是一个运行在 Java8 之上满足 Reactice 规范的响应式框架，它提供了一组响应式风格的 API。

Reactor 有两个核心类： Flux<T> 和 Mono<T>，这两个类都实现 Publisher 接口。

• Flux 类似 RxJava 的 Observable，它可以触发零到多个事件，并根据实际情况结束处理或触发错误。

• Mono 最多只触发一个事件，所以可以把 Mono 用于在异步任务完成时发出通知。

Flux 和 Mono 都是数据流的发布者，使用 Flux 和 Mono 都可以发出三种数据信号：元素值，错误信号，完成信号；错误信号和完成信号都代表终止信号，终止信号用于告诉订阅者数据流结束了，错误信号终止数据流同时把错误信息传递给订阅者。

三种信号的特点：

• 错误信号和完成信号都是终止信号，不能共存
• 如果没有发送任何元素值，而是直接发送错误或者完成信号，表示是空数据流
• 如果没有错误信号，也没有完成信号，表示是无限数据流

引入依赖

        <dependency>
            <groupId>org.projectreactor</groupId>
            <artifactId>reactor-core</artifactId>
            <version>1.1.6.RELEASE</version>
        </dependency>

just 和 subscribe方法

just()：创建Flux序列，并声明指定数据流

subscribe()：订阅Flux序列，只有进行订阅后才回触发数据流，不订阅就什么都不会发生

public class TestReactor {
    public static void main(String[] args) {
        //just()：创建Flux序列，并声明数据流，
        Flux<Integer> integerFlux = Flux.just(1, 2, 3, 4);//整形
        //subscribe()：订阅Flux序列，只有进行订阅后才回触发数据流，不订阅就什么都不会发生
        integerFlux.subscribe(System.out::println);
        
        Flux<String> stringFlux = Flux.just("hello", "world");//字符串
        stringFlux.subscribe(System.out::println);
        
        //fromArray(),fromIterable()和fromStream()：可以从一个数组、Iterable 对象或Stream 对象中创建Flux序列
        Integer[] array = {1,2,3,4};
        Flux.fromArray(array).subscribe(System.out::println);
        
        List<Integer> integers = Arrays.asList(array);
        Flux.fromIterable(integers).subscribe(System.out::println);
        
        Stream<Integer> stream = integers.stream();
        Flux.fromStream(stream).subscribe(System.out::println);
    }
}

启动测试：

这篇好文章是转载于：学新通技术网