【深入解析spring cloud gateway】08 Reactor 知识扫盲

一、响应式编程概述

1.1 背景知识

为了应对高并发服务器端开发场景，在2009 年，微软提出了一个更优雅地实现异步编程的方式——Reactive Programming，我们称之为响应式编程。随后，Netflix 和LightBend 公司提供了RxJava 和Akka Stream 等技术，使得Java 平台也有了能够实现响应式编程的框架。
在2017 年9 月28 日，Spring 5 正式发布。Spring 5 发布最大的意义在于，它将响应式编程技术的普及向前推进了一大步。而同时，作为在背后支持Spring 5 响应式编程的框架Spring Reactor，也进入了里程碑式的3.1.0 版本。

1.2 什么是响应式编程

响应式编程是一种面向数据流和变化传播的编程范式。这意味着可以在编程语言中很方便地表达静态或动态的数据流，而相关的计算模型会自动将变化的值通过数据流进行传播。
响应式编程基于reactor（Reactor 是一个运行在 Java8 之上的响应式框架）的思想，当你做一个带有一定延迟的才能够返回的io操作时，不会阻塞，而是立刻返回一个流，并且订阅这个流，当这个流上产生了返回数据，可以立刻得到通知并调用回调函数处理数据。
电子表格程序就是响应式编程的一个例子。单元格可以包含字面值或类似"=B1+C1"的公式，而包含公式的单元格的值会依据其他单元格的值的变化而变化。
响应式传播核心特点之一：变化传播：一个单元格变化之后，会像多米诺骨牌一样，导致直接和间接引用它的其他单元格均发生相应变化。

1.3 基于 Reactor 实现

Reactor 是一个运行在 Java8 之上满足 Reactice 规范的响应式框架，它提供了一组响应式风格的 API。
Reactor 有两个核心类： Flux 和 Mono，这两个类都实现 Publisher 接口。
Flux 类似 RxJava 的 Observable，它可以触发零到多个事件，并根据实际情况结束处理或触发错误。
Mono 最多只触发一个事件，所以可以把 Mono 用于在异步任务完成时发出通知。
Flux 和 Mono 都是数据流的发布者，使用 Flux 和 Mono 都可以发出三种数据信号：元素值，错误信号，完成信号；错误信号和完成信号都代表终止信号，终止信号用于告诉订阅者数据流结束了，错误信号终止数据流同时把错误信息传递给订阅者。
三种信号的特点：
错误信号和完成信号都是终止信号，不能共存
如果没有发送任何元素值，而是直接发送错误或者完成信号，表示是空数据流
如果没有错误信号，也没有完成信号，表示是无限数据流

Mono 原理图如下：
在这里插入图片描述

Flux原理图如下：
在这里插入图片描述

结合上面两个图，发现Mono和Flux非常相似。只是Mono只接收一个元素，而Flux接收多个元素

二、示例代码

2.1 Mono

package com.reactor.demo;import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.publisher.Mono;import java.time.Duration;
import java.util.Optional;
import java.util.function.Consumer;
import java.util.function.Function;@Slf4j
public class MonoTest {@Testpublic void test1() {//just用法Mono.just("hello world").subscribe(System.out::println);//runnable创建monoMono<Void> sinkMono = Mono.fromRunnable(() -> System.out.println("runnable"));//这句不会输出sinkMono.doOnNext(unused -> System.out.println("void success"));//这句也不会输出sinkMono.subscribe(o -> System.out.println("void result" + o));//创建一个不包含任何元素，只发布结束消息的序列。，这里的hello empty是不会输出的。Mono.empty()//输出“empty的入参是null”.doOnSuccess(o -> System.out.println("empty的入参是" + o))//这句不会输出.subscribe(o -> System.out.println("hello empty"));//empty里面至少还有一个结束消息，而never则是真的啥都没有。"never的入参是"不会输出 ，这里的hello never也不会输出Mono.never().doOnSuccess(o -> System.out.println("never的入参是" + o)).subscribe(o -> System.out.println("hello never"));}@Testpublic void test2() {//传入supplierMono.fromSupplier(() -> "Hello supplier").subscribe(System.out::println);//传入optionalMono.justOrEmpty(Optional.of("Hello optional")).subscribe(System.out::println);//通过sink来创建一个正常执行的MonoMono.create(sink -> sink.success("Hello sink")).subscribe(System.out::println);//通过sink来创建一个抛出异常的MonoMono.create(sink -> sink.error(new RuntimeException("sink error"))).subscribe(System.out::println);//defer的入参实际上是一个Mono工厂Mono.defer(() -> Mono.just("hello defer")).subscribe(System.out::println);}@Testpublic void test3() {//callable，有返回值Mono.fromCallable(() -> "callable").subscribe(System.out::println);//runnable无返回值Mono<Void> mono = Mono.fromRunnable(() -> System.out.println("run"));//下面的hello runnable是不会输出的。因为subscribe一个Mono<Void>,不会产生任何结果mono.subscribe(o -> System.out.println("hello runnable"));}@Testpublic void test4() {//延迟3秒输出Mono.delay(Duration.ofSeconds(3)).doOnNext(new Consumer<Long>() {@Overridepublic void accept(Long aLong) {System.out.println(aLong);}}).block();}@Testpublic void test5() {//直接输出了异常Mono.error(new RuntimeException("这是一个异常")).subscribe(new Consumer<Object>() {@Overridepublic void accept(Object o) {System.out.println("error:" + o);}});Mono.defer(() -> {return Mono.error(new RuntimeException("这是第二个异常"));}).subscribe(new Consumer<Object>() {@Overridepublic void accept(Object o) {System.out.println("defer error:" + o);}});}@Testpublic void test6() {//通过map可以对元素进行转换Mono.just("just one").map(new Function<String, Integer>() {@Overridepublic Integer apply(String s) {return 1;}}).doOnNext(new Consumer<Integer>() {@Overridepublic void accept(Integer integer) {System.out.println("转换后的结果：" + integer);}}).subscribe();}
}

2.1 Flux

package com.reactor.demo;import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;import java.time.Duration;
import java.util.Arrays;
import java.util.function.Consumer;public class FluxTest {/*** 基本用法*/@Testpublic void test1() {//通过just传入可变的参数，依次输出Flux.just("hello", "world", "just").doOnNext(System.out::println).doOnComplete(() -> System.out.println("just over")).subscribe();//传入一个范围Flux.range(100, 10).doOnNext(System.out::println).doOnComplete(() -> System.out.println("OK")).subscribe();//传入listFlux.fromIterable(Arrays.asList("01", "02", "03")).doOnNext(System.out::println).subscribe();//传入一个数组Flux.fromArray(new Object[]{"obj1", "obj2"}).doOnNext(System.out::println).subscribe();}/*** 处理空值*/@Testpublic void testEmpty() {//如果序列是个空的，就给个默认值Flux.empty().defaultIfEmpty(1).doOnNext(System.out::println).subscribe();//如果序列是空的，就用新序列代替Flux.empty().switchIfEmpty(Mono.just("100")).doOnNext(System.out::println).subscribe();}/*** 序列在执行时的一些监听方法doOnXXXX*/@Testpublic void testDoOn() {System.out.println("----------");Flux.range(100, 10).doOnNext(System.out::println).doOnComplete(() -> System.out.println("OK"));System.out.println("----------");Flux.range(100, 10).doFirst(() -> System.out.println("第一个执行开始")).subscribe();System.out.println("----------");Flux.range(100, 10).doFinally(it -> System.out.println("终止信号的类型为" + it.name())).subscribe();System.out.println("----------");Flux.range(100, 10).doOnSubscribe(it -> System.out.println("该序列已被订阅")).subscribe();System.out.println("----------");Flux.range(100, 10).doOnRequest(value -> System.out.println("doOnRequest:" + value)).subscribe();//在完成或者error时，也就是序列终止时执行runnableSystem.out.println("----------");Flux.range(100, 10).doOnTerminate(() -> System.out.println("doOnTerminate")).subscribe();//doOnEach每次向下游传播，都会得到一个信号类型，可以根据该信号类型执行一些操作System.out.println("----------");Flux.range(100, 10).doOnEach(it -> System.out.println("doOnEach:" + it)).subscribe();}/*** filter用法*/@Testpublic void testFilter() {System.out.println("----------");//将上游的数据进行类型判断，符合该类型的数据将流向下游Flux.just(new Object(), "Hello", 1).ofType(String.class).doOnNext(System.out::println).doOnComplete(() -> System.out.println("过滤String示例")).subscribe();System.out.println("----------");//过滤数据Flux.range(100, 10).filter(it -> it > 105).doOnComplete(() -> System.out.println("取出大于105示例")).subscribe();System.out.println("----------");//将重复数据过滤,重复数据在整个序列中只保留一个Flux.range(100, 10).concatWith(Flux.just(100, 100, 100)).distinct().doOnNext(System.out::println).doOnComplete(() -> System.out.println("去除重复数字示例")).subscribe();System.out.println("----------");//将后来的重复数据过滤，如下，第二个flux拼接到第一个序列时，只会把第二个元素本身的重复元素过滤Flux.range(100, 10).concatWith(Flux.just(100, 100, 100)).distinctUntilChanged().doOnNext(System.out::println).doOnComplete(() -> System.out.println("将后来的重复数据过滤")).subscribe();System.out.println("----------");//在序列的开始获取5个元素，// limitRequest为true时，则不管该序列会发射多少元素，该参数会向上传递背压，则上游序列只会发出设定的5个元素//为false时，则不控制上有元素可以发出N个元素Flux.range(100, 10).take(5, false).doOnComplete(() -> System.out.println("在序列的开始获取5个元素")).subscribe();System.out.println("----------");//参数为时间单位，意味着take获取元素，只会在该时间限制内获取。Flux.range(100, 10).take(Duration.ofSeconds(10)).doOnNext(new Consumer<Integer>() {@Overridepublic void accept(Integer integer) {System.out.println("当前时间戳为：" + System.currentTimeMillis() + "，数字为：" + integer);}}).doOnComplete(() -> System.out.println("在指定时间内获取元素")).subscribe(System.out::println);System.out.println("----------");//获取最后的N位元素Flux.range(100, 10).takeLast(2).doOnComplete(() -> System.out.println("获取最后的2位元素")).subscribe(System.out::println);System.out.println("----------");//获取元素，知道符合条件后停止向下游发送数据,包括条件本身，也就是当it>105的元素也会被发布至下游Flux.range(100, 10).takeUntil(it -> it > 105).doOnComplete(() -> System.out.println("一直取数，直到大于105结束")).subscribe(System.out::println);System.out.println("----------");//获取元素，当元素符合该断言时，如果不符合直接终止,不包含条件本身Flux.range(100, 10).takeWhile(it -> it < 105).doOnComplete(() -> System.out.println("取出小于105示例")).subscribe(System.out::println);System.out.println("----------");//获取指定某个位置的一个元素Flux.range(100, 10).elementAt(0).doOnSuccess(new Consumer<Integer>() {@Overridepublic void accept(Integer i) {System.out.println("获取指定某个位置的一个元素:" + i);}}).subscribe();System.out.println("----------");//获取最后一个元素，last()如果为空则抛出异常，last(1)如果为空则发出默认值Flux.range(100, 10).takeWhile(it -> it > 105).last(1).subscribe(System.out::println);System.out.println("----------");//跳至第几秒开始执行Flux.range(100, 10).skip(Duration.ofSeconds(5)).subscribe(System.out::println);System.out.println("----------");//跳至第几个元素开始执行Flux.range(100, 10).skip(5).subscribe(System.out::println);System.out.println("----------");//从开始跳到最后第N个元素结束Flux.range(100, 10).skipLast(5).subscribe(System.out::println);System.out.println("----------");//跳至满足条件的地方开始执行,从第一个元素开始，知道满足条件，开始发送至下游Flux.range(100, 10).skipUntil(it -> it > 105).subscribe(System.out::println);System.out.println("----------");//每隔一段时间抽取样本数(取在这个时间的最后一个元素)，如果相隔实现大于序列的执行时间，则去最后一元素Flux.range(100, 100000000).sample(Duration.ofMillis(100)).subscribe(System.out::println);System.out.println("----------");//每隔一段时间抽取样本数(取在这个时间的第一个元素)，如果相隔实现大于序列的执行时间，则取第一个元素Flux.range(100, 10).sampleFirst(Duration.ofMillis(100)).subscribe(System.out::println);System.out.println("----------");//只获取一个元素，single()如果为空或者超多一个，抛出异常，single(1)如果为空返回默认值，如果多个抛出异常,singleOrEmpty()可以允许为空Flux.range(100, 10).single(1).subscribe(System.out::println);}/*** 当被订阅后如果发生异常，则stream会停止运行* 此时可以通过处理error来决定如何处理异常* 可以将异常跳过、将异常替换等*/@Testpublic void testErrorHandle() {System.out.println("----------");Flux.just(1, 2, 3, 0, 5, 4).map(it -> {it = 100 / it;return it;})//报错后返回，并停止运行.onErrorResume(e -> {return Mono.just(10000);}).doFinally(type -> {System.out.println(type);}).subscribe(System.out::println);System.out.println("----------");Flux.just(1, 2, 3).doOnNext(new Consumer<Integer>() {@Overridepublic void accept(Integer integer) {System.out.println(integer);if (integer == 2) {throw new RuntimeException("触发异常");}}}).doOnError(new Consumer<Throwable>() {@Overridepublic void accept(Throwable throwable) {System.out.println("doOnError：" + throwable.getMessage());}}).subscribe();System.out.println("----------");Flux.just(1, 2, 3, 0, 5, 4).map(it -> {it = 100 / it;return it;})//报错后继续运行，并执行相关操作.onErrorContinue((e, it) -> {System.out.println(e.getMessage());}).doFinally(type -> {System.out.println(type);}).subscribe(System.out::println);}@Testpublic void flatMapTest() {//输出50，100Flux.just(5, 10).flatMap(x -> Flux.just(x * 10)).toStream().forEach(System.out::println);}
}

参考文章

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/122838.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

【K 均值聚类】02/5：简介

【K 均值聚类】02/5：简介

一、说明 k-mean算法是一种聚类算法&#xff0c;它的主要思想是基于数据点之间的距离进行聚类。K-means聚类是一种无监督的机器学习算法。让我们再解释一下这句话。聚类分析的目标是将数据划分为同类聚类。每个聚类中的点彼此之间比其他聚类中的点更相似。 无监督机器学习是在没…
阅读更多...
Ubuntu18中NVIDIA，cuda,cudnn,pytorch安装

Ubuntu18中NVIDIA，cuda,cudnn,pytorch安装

注意&#xff1a;nvidia驱动和cuda,cudnn,pytroch,python的对应关系 linux安装pytorch&#xff08;包括cuda与cudnn&#xff09;_linux清华园按照pytorch1.12_BryceRui的博客-CSDN博客 安装流程&#xff1a;安装cuda&#xff08;包括nvidia驱动&#xff09; cudnn python安装…
阅读更多...
【蒸汽冷凝器型号和PI控制】具有PID控制的蒸汽冷凝器的动力学模型（MatlabSimulink）

【蒸汽冷凝器型号和PI控制】具有PID控制的蒸汽冷凝器的动力学模型（MatlabSimulink）

&#x1f4a5;&#x1f4a5;&#x1f49e;&#x1f49e;欢迎来到本博客❤️❤️&#x1f4a5;&#x1f4a5; &#x1f3c6;博主优势&#xff1a;&#x1f31e;&#x1f31e;&#x1f31e;博客内容尽量做到思维缜密&#xff0c;逻辑清晰&#xff0c;为了方便读者。 ⛳️座右铭&a…
阅读更多...
【爬虫】7.4. 字体反爬案例分析与爬取实战

【爬虫】7.4. 字体反爬案例分析与爬取实战

字体反爬案例分析与爬取实战 文章目录 字体反爬案例分析与爬取实战1. 案例介绍2. 案例分析3. 爬取 本节来分析一个反爬案例&#xff0c;该案例将真实的数据隐藏到字体文件里&#xff0c;即使我们获取了页面源代码&#xff0c;也无法直接提取数据的真实值。 1. 案例介绍 案例网…
阅读更多...
计算机网络概述

计算机网络概述

目录 一、计算机网络的作用及互联网概述 1.1计算机网络在信息时代中的作用 1.2基本概念 1.3互联网基础架构发展三个阶段 1.4互联网的标准化工作 二、互联网的组成 2.1互联网组成 2.2互联网的边缘部分 2.3互联网的核心部分 三、计算机网络的类别 3.1计算机网络的定义:…
阅读更多...
加强版python连接飞书通知——本地电脑PC端通过网页链接打开本地已安装软件（调用注册表形式，以漏洞扫描工具AppScan为例）

加强版python连接飞书通知——本地电脑PC端通过网页链接打开本地已安装软件（调用注册表形式，以漏洞扫描工具AppScan为例）

前言 如果你想要通过超链接来打开本地应用，那么你首先你需要将你的应用添入windows注册表中（这样网页就可以通过指定代号来调用程序），由于安全性的原因所以网页无法直接通过输入绝对路径来调用本地文件。 一、通过创建reg文件自动配置注册表 创建文本文档，使用记事本打开…
阅读更多...
蓝桥杯打卡Day3

蓝桥杯打卡Day3

文章目录 吃糖果递推数列 一、吃糖果IO链接 本题思路:本题题意就是斐波那契数列&#xff01; #include <bits/stdc.h>typedef uint64_t i64;i64 f(i64 n) {if(n1) return 1;if(n2) return 2;return f(n-1)f(n-2); }signed main() {std::ios::sync_with_stdio(false);s…
阅读更多...
苍穹外卖集成 Apache POI Java实现Excel文件的读写下载

苍穹外卖集成 Apache POI Java实现Excel文件的读写下载

苍穹外卖 day12 Echats 营业台数据可视化整合_软工菜鸡的博客-CSDN博客 Apache POI - the Java API for Microsoft Documents Project News 16 September 2022 - POI 5.2.3 available The Apache POI team is pleased to announce the release of 5.2.3. Several dependencies …
阅读更多...
AJAX学习笔记8 跨域问题及解决方案

AJAX学习笔记8 跨域问题及解决方案

AJAX学习笔记7 AJAX实现省市联动_biubiubiu0706的博客-CSDN博客 跨域:指一个域名的网页去请求另外一个域名资源.比如百度页面去请求京东页面资源. 同源与不同源三要素:协议,域名,端口 协议一致,域名一致,端口一致.才算是同源.其他一律不同源 新建项目测试: 1.window.open();…
阅读更多...
HTML emoji整理 表情符号

HTML emoji整理 表情符号

<!DOCTYPE html> <html lang"en"><head><meta charset"UTF-8" /><title>测试</title></head><body><div style"font-size: 50px;">&#128276</div><script>let count 0d…
阅读更多...
Yolov5的tensorRT加速(python)

Yolov5的tensorRT加速(python)

地址&#xff1a;https://github.com/wang-xinyu/tensorrtx/tree/master/yolov5 下载yolov5代码 方法一&#xff1a;使用torch2trt 安装torch2trt与tensorRT 参考博客&#xff1a;https://blog.csdn.net/dou3516/article/details/124538557 先从github拉取torch2trt源码 ht…
阅读更多...
代码随想录算法训练营第二十四天｜理论基础 77. 组合

代码随想录算法训练营第二十四天｜理论基础 77. 组合

理论基础 其实在讲解二叉树的时候&#xff0c;就给大家介绍过回溯&#xff0c;这次正式开启回溯算法&#xff0c;大家可以先看视频&#xff0c;对回溯算法有一个整体的了解。 题目链接/文章讲解&#xff1a;代码随想录 视频讲解&#xff1a;带你学透回溯算法&#xff08;理论篇…
阅读更多...
函数栈帧(详解)

函数栈帧(详解)

一、前言&#xff1a; 环境&#xff1a;X86Vs2013 我们C语言学习过程中是否遇到过如下问题或者疑惑&#xff1a; 1、局部变量是如何创建的&#xff1f; 2、为什么局部变量的值是随机值&#xff1f; 3、函数是怎么传参的&#xff1f;传参的顺序是怎样的&#xff1f; 4、形…
阅读更多...
HarmonyOS/OpenHarmony(Stage模型）应用开发单一手势（三）

HarmonyOS/OpenHarmony(Stage模型）应用开发单一手势（三）

五、旋转手势&#xff08;RotationGesture&#xff09; RotationGesture(value?:{fingers?:number; angle?:number}) 旋转手势用于触发旋转手势事件&#xff0c;触发旋转手势的最少手指数量为2指&#xff0c;最大为5指&#xff0c;最小改变度数为1度&#xff0c;拥有两个可…
阅读更多...
3D异常检测论文笔记 | Shape-Guided Dual-Memory Learning for 3D Anomaly Detection

3D异常检测论文笔记 | Shape-Guided Dual-Memory Learning for 3D Anomaly Detection

文章目录 摘要一、介绍三、方法3.1. 形状引导专家学习3.2. Shape-Guided推理 摘要 我们提出了一个形状引导的专家学习框架来解决无监督的三维异常检测问题。我们的方法是建立在两个专门的专家模型的有效性和他们的协同从颜色和形状模态定位异常区域。第一个专家利用几何信息通…
阅读更多...
机器学习笔记：node2vec（论文笔记：node2vec: Scalable Feature Learning for Networks）

机器学习笔记：node2vec（论文笔记：node2vec: Scalable Feature Learning for Networks）

2016 KDD 1 intro 利用graph上的节点相似性&#xff0c;对这些节点进行embedding 同质性&#xff1a;节点和其周围节点的embedding比较相似 蓝色节点和其周围的节点结构等价性 结构相近的点embedding相近 比如蓝色节点&#xff0c;都处于多个簇的连接处 2 随机游走 2.1 介绍…
阅读更多...
『C语言进阶』指针进阶(一)

『C语言进阶』指针进阶(一)

&#x1f525;博客主页&#xff1a; 小羊失眠啦 &#x1f516;系列专栏&#xff1a; C语言 &#x1f325;️每日语录&#xff1a;无论你怎么选&#xff0c;都难免会有遗憾。 ❤️感谢大家点赞&#x1f44d;收藏⭐评论✍️ 前言 在C语言初阶中&#xff0c;我们对指针有了一定的…
阅读更多...
《机器人学一(Robotics(1))》_台大林沛群 第 5 周【机械手臂 轨迹规划】 Quiz 5

《机器人学一(Robotics(1))》_台大林沛群 第 5 周【机械手臂 轨迹规划】 Quiz 5

我又行了&#xff01;&#x1f923; 求解的 位置 可能会有 变动&#xff0c;根据求得的A填写相应值即可。注意看题目。 coursera链接 文章目录 第1题 Cartesian space求解 题1-3 的 Python 代码 第2题第3题第4题 Joint space求解 题4-6 的 Python 代码 第5题第6题其它可参考代…
阅读更多...
编写软件检测报告有哪些注意事项?软件检测报告获取

编写软件检测报告有哪些注意事项?软件检测报告获取

软件检测报告是指把测试的过程和结果写成文档&#xff0c;对发现的问题和缺陷进行分析&#xff0c;为纠正软件的存在的质量问题提供依据&#xff0c;同时为软件验收和交付打下基础。 一、编写软件检测报告的注意事项 1、报告的结构要合理和清晰。应该按照一定的逻辑顺序&…
阅读更多...
解决 Spring Boot 与 springfox 的 NullPointerException 问题

解决 Spring Boot 与 springfox 的 NullPointerException 问题

&#x1f337;&#x1f341; 博主猫头虎&#xff08;&#x1f405;&#x1f43e;&#xff09;带您 Go to New World✨&#x1f341; &#x1f984; 博客首页——&#x1f405;&#x1f43e;猫头虎的博客&#x1f390; &#x1f433; 《面试题大全专栏》 &#x1f995; 文章图文…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023