使用项目反应器 mergeWith() 运算符以实现“if/elseif/else"分支逻辑
我正在尝试使用 project reactor mergeWith 运算符来实现 if/elseif/else 分支逻辑，如下所述:RxJS，If-Else 运算符在哪里. 提供的示例是用 RxJS 编写的，但基本思想保持不变. 基本上的想法是在 3 个 monos/publishers(因此有 3 个不同的谓词)上使用 filter 运算符并合并 3 个 monos如下(这里当然是
..
project-reactor相关内容
使用 application/stream+json 时 WebFlux 不发送数据
我有一个反应式核心 WebClient 可以发布到给定的端点.负载是 Job 对象的流量，内容类型是 application/stream+jsonFlux jobFlux = Flux.just(new Job()); Monoresponse = localEP.post().uri( "/dev/job" ).contentType(MediaType.APPLICATION_STREAM
..
将 Mono 与 Flux 结合
我想创建一个服务，将来自两个反应源的结果结合起来.一个是生产 Mono，另一个是生产 Flux.对于合并，对于每个发射的通量，我需要相同的单声道值. 现在我有这样的东西 Flux.zip(service1.getConfig(),//产生通量service2.getContext()//产生单声道.cache().repeat()) 这给了我我需要的东西， service2 只调用
..
如何使用 Spring AOP 和 WebFlux 获取从 joinPoint.proceed() 返回的对象
我有一个带有 @Around 注释的简单方面(见下文).当应用程序不使用反应式范式时，这方面起作用.但是当应用程序返回 Mono 或 Flux 时无法正常工作. 我需要获取方法返回的对象来生成一个JSON对象来用作日志、生成事件等 这是我在非响应类中工作的代码: @Around("@annotation(simpleEvent)")公共对象 logExecutionTime(Proc
..
Webflux WebClient 异步请求和处理 Mono
我是 webflux 的新手，无法找到合适的材料来继续实施. 我想发出请求并异步处理响应.在这种情况下，服务调用大约需要 8-10 毫秒来响应，因此我们发出请求并继续做其他工作，并在需要进一步处理时寻找响应. Mono>resp = webClient.post().uri("/{type}",isCustomerPricing ? "customer" : "profile").body
..
rxJava buffer() 与时间背压
buffer 操作符的版本不按 JavaDoc 进行时间荣誉背压: http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#buffer-int- 然而，任何涉及基于时间的缓冲区的buffer版本都不支持背压，就像这个 http://reactivex.io/RxJava/2.x/javadoc/io/r
..
变换与变换延迟
transform 和有什么区别?transformDeferred 在项目反应堆通量中. 好的例子会有帮助. https://projectreactor.io/docs/core/release/reference/index.html#advanced-mutualizing-operator-usage 解决方案 大多数时候，Flux 是“懒惰的":你声明一个处理管道，
..
如何从阻塞队列创建反应堆 Flux?
我正在尝试实现从 BlockingQueue 创建的 reactor Flux 但不确定哪个运算符最适合我的用例? 我正在创建一个流式 REST 端点，其中响应是 Flux，它需要不断地从 BlockingQueue 发出消息作为对 GET REST 调用的响应. 我已经尝试过论坛和文档，但只能找到从可迭代集合或响应式数据源启动的 Flux，但没有来自任何 BlockingQueue
..
如何使 AuditorAware 与 Spring Data Mongo Reactive 一起工作
Spring Security 5 提供了一个 ReactiveSecurityContextHolder 来从 Reactive 中获取 SecurityContext上下文，但是当我想实现 AuditorAware 并自动获得试听工作时，但它不起作用.目前我找不到 AuditorAware 的 Reactive 变体. @Beanpublic AuditorAware审计员(){返回 ()
..
使用 StepVerifier 对 Flux.take(Duration duration) 进行单元测试
我使用的是 Spring Reactor Core 3.0.6 并且我有一个返回 Flux 的方法: public Flux创建流量(){return Flux.,String>generate(/* 为了清楚起见省略了生成器 */).take(Duration.ofSeconds(10).flatMap(Flux::fromIterable);} 生成器函数调用分页的 R
..
Spring-Webflux:在没有 block() 的情况下从 Mono 中提取对象
我是新的 Spring Webflux.我正在编写一个简单的 api，它调用另一个 api 并返回响应.我的问题是我的 api 接受的请求类型与外部 api 不同.我必须将传入的请求转换为发送到外部 api.我正在使用 Mono 来接收对我的 api 的请求，但是在没有转换到另一个对象时遇到了麻烦块(). 输入 路由器 @Configuration@EnableWebFlux公共类
..
Spring Reactive Webclient 的请求级背压?
这个类似于如何做akka-http request-level背压? 但对于 Spring 回声系统. 我正在考虑如何在以响应方式使用 Spring WebClient 时为 HTTP 客户端实现背压.对我来说，似乎要走的路是让 WebClient 意识到 HTTP 语义并对例如应用背压.状态“429 - 请求过多".我没有找到任何关于此的文档，这让我有点怀疑这是否是可行的方法. 问题
..
异常处理反应式
如何使用 Mono.error() 但附加请求返回的正文中的信息? 是否有一个响应式对象扩展了 Throwable 并接受 Mono/Flux 对象，因此抛出的错误将等待主体被解释? 或者有没有办法在现有的 Mono 对象上添加某种“标志"以使其立即失败(规避对 Throwable 的要求) 以下示例场景: import org.springframework.web.reacti
..
在反应堆中流动多个物体的有效/正确方法是什么
我是响应式编程的新手，我正在尝试构建一个接近真实的示例. 当您看到反应堆教程时，它们会向您展示非常简单的示例，例如. return userRepository.findById(1); 或类似处理通量之类的东西打破“棕色小狐狸"；字符串并找到唯一的字母等.但大多数这些教程都坚持单个对象，不幸的是，我无法找到任何指导线或教程，它们显示并排示例以先在命令式然后在反应式中键入相同的代码，这就
..
带有 WebFlux 的 Spring Boot 总是在测试中抛出 403 状态
非常感谢您查看我的问题)我有一些奇怪的主题:我的春季启动测试不起作用.他们成功启动但在向任何控制器发出请求时总是抛出 403 HTTP 状态我有一些具有下一个依赖项的项目: buildscript {ext.kotlin_version = '1.3.71'依赖{类路径“org.jetbrains.kotlin:kotlin-gradle-plugin:1.3.71"类路径“org.spring
..
Project Reactor 和 Java 内存模型
我试图了解 Project reactor 为应用程序代码提供的数据可见性保证.例如我希望下面的代码会失败，但经过一百万次迭代后不会失败.我正在更改线程 A 上典型 POJO 的状态并从线程 B 读取它. Reactor 是否保证 POJO 更改在线程间可见? public class Main {公共静态无效主(字符串 [] args){整数结果 = Flux.range(1, 1_000_0
..
将一个通量分成两个通量 - 头和尾
我想将一个通量拆分为两个通量，其中第一个通量具有原始通量的第一项，第二个通量将获得其余项. 在每个通量上应用自定义转换 myLogic 后，我想将它们组合成一个通量，保留原始通量的顺序. 示例: S:学生 S':申请myLogic 后的学生 发射通量:s1 ->s2 ->s3 ->s4 第一个分裂通量:s1' =>myLogic 第二个分裂通量:s2' ->s3' ->
..
项目反应堆.Mono.map() 与 Mono.flatMap()
就Mono 而言，它们之间的主要区别是什么?从文档中，我读到 flatMap 是异步的，而 map 是同步的.但这对我来说并没有真正意义 b/c Mono 是关于并行性的，这一点是不可理解的.有人可以用更容易理解的方式重新表述它吗? 然后在 flatMap 的文档中说明 (https://projectreactor.io/docs/core/release/api/reactor/core
..
在 Spring Webflux 中执行阻塞 JDBC 调用
我将 Spring Webflux 与 Spring 数据 jpa 一起使用，使用 PostgreSql 作为后端数据库.我不想在进行诸如 find 和 save 之类的数据库调用时阻塞主线程.为了实现相同的目的，我在 Controller 类和 jdbcScheduler 服务类中有一个主调度程序. 我定义它们的方式是: @Configuration@EnableJpaAuditing公
..
如何使用 Reactor (Spring WebClient) 进行重复调用?
我使用 Reactor (Spring5 WebClient) 作为我的反应式编程 API.我有 2 个 REST 端点要调用.第一个的结果将是第二个的参数.对于第二个 API，它将返回一个带有“hasMore"值的结果.如果该值为 true，我应该更改分页参数并再次调用第二个 API.演示代码如下: client.getApi1().map(r -> r.getResult()).flatM
..