我使用轮询方法定期获取数据。新数据随时可能到来。我想向我的客户公开一个反应式接口。所以，我想创建一个发布器(Flux?)这将在新数据可用时发布新数据，并通知订阅者。我该怎么做？我看到的Flux的所有示例都是针对数据已知/可用的情况。实际上，我想要的是基于队列的Flux，当我的轮询线程发现新数据时，它可以继续填充队列。
浏览 27提问于2019-06-25得票数 5
1回答
反应堆3有两种主要数据类型，它们都是反应性流发布者。reactor.core.publisher.Mono<T>
我理解Mono是0或1元素流与Flux是0或N元素流之间的区别。既然Mono和Flush都在实现org.reactivestreams.Publisher<T>，为什么我们需要这两种类型，为什么不直接使用F
浏览 0提问于2018-01-09得票数 1
回答已采纳
2回答
我的usecase是创建如下所示的反应性端点： ServicePoller poller =So that I can access collection later on without polling again */
这会在数据可用时立即将数据发回前端，然而，我的第二个用例是在数据到达时将数据集中到一个单独的集合中在从原始流量返回控制器之前，我尝试订阅
浏览 1提问于2018-06-22得票数 1
回答已采纳
1回答
return from(someReactiveInboundChannelAdapter) }
但这是行不通的，因为框架无法识别flux是一个Flux<?
浏览 5提问于2022-04-08得票数 1
回答已采纳
1回答
flatMap()内部的单声道是具有复杂逻辑的复杂反应流。有没有办法将Mono保存为可以传递给父流的变量？换句话说，就像这样： final Function<Integer, Publisher<String>> monoPublisher = Mono.something()然而，这仍然会将Mono的初始化推迟到实际的订阅，并且会为父流中的每个元素创建新的Mono。我熟悉Mono.create()，但在这里我不知道如何让它工作，除非创建我自己的Publi
浏览 26提问于2020-10-27得票数 1
1回答
org.springframework.web.reactive.function.server.ServerRequest;import reactor.core.publisher.Mono
浏览 9提问于2020-03-16得票数 45
2回答
我有一个Flux流，它从数据库读取对象。我希望处理函数在获得给定对象的ID的Redis锁之后执行，并在处理之后释放锁(如果处理函数抛出错误)。在Flux中创建这样的流最简单的方法是什么？
以下是我尝试使用transform进行此操作失败的一些代码。我希望这是尽可能透明的其他流，不需要单独的连接锁和解锁功能，只是一个附件，可以做“锁-进程-解锁”。private <T> Function<Flux</e
浏览 11提问于2022-04-11得票数 0
我的理解是，Mono<List<T>>是一个同步的Flux<T>，而Flux不能是rest响应。
我说的对吗？如果不是，Mono<List<T>>和Flux<T>之间有什么不同，或者在某些地方，Flux可以是rest响应？
浏览 3提问于2018-10-22得票数 15
回答已采纳
2回答
我试图在Flux/Publisher上执行以下操作，这些操作只能读取一次(考虑数据库结果，可以读取一次)。但是，这个问题具有足够的通用性，可以在没有反应堆知识的情况下在函数式编程环境中回答。distinctAndHasElement(4, Flux.just(1,2,3,3,4,4,5));
Mono<Pair<Long, Boolean>> distinctAndHasElement(int toCheck, Flux<Integer> intsFlux
浏览 9提问于2021-06-08得票数 1
回答已采纳
我创建了这样一个WebSocketHandler实现：public Mono<Void> handle(WebSocketSession session) {
var publisher = flux.map( o -> {
try {但是，我想对来自websocket的请求做出反应，告诉servi
浏览 0提问于2019-03-06得票数 3
回答已采纳
我们有一个多组分应用程序，它提供了组件之间的反应性流API。一些组件使用Akka流实现，另一些组件使用例如反应堆。Publisher<String> stringPublisher = Flux.fromIterable(Sink.asPublis
浏览 0提问于2019-04-05得票数 1
回答已采纳
在ProjectReactor或反应性流中，在订阅().之前，不会发生任何事情。除非有人订阅了，否则将不会发生反应性流数据流，但我看到对于所有REST API，如find、save和inserts并不是显式调用订阅，而是在生产者和订阅者之间流动数据。repository) { } public Flu
浏览 1提问于2018-06-11得票数 10
回答已采纳
public Flux<String> mqttChannel() { return mqttFlux;我该怎么做呢？我的Flux不能以类似的方式获得它的输入吗--以某种方式将它传递给IntegrationFlows处理函数？MQTT示例代码取自
浏览 1提问于2020-10-09得票数 1
我有以下示例代码： Flux<Integer> range = Flux.range(0, 10);Flux<Integer> flux3 = flux1
浏览 57提问于2019-04-09得票数 1
2回答
因此，我添加了一个额外的心跳流，并将其与数据连接流合并。由于我现在在数据连接上返回字符串，所以我可以简单地从心跳返回一个字符串，因此从流的合并返回一个Flux<String>： }
但是，如果我的数据流返
浏览 1提问于2019-11-25得票数 1
回答已采纳
1回答
我想用反应堆的通量提供一些数据。由于提供这些数据可能需要很长时间，所以我决定引入ping机制(例如，保持tcp连接活动而不超时)。问题是：如何在Junit5测试中测试这个ping机制，这样就不会花费太多时间(例如几秒钟)
浏览 5提问于2021-01-15得票数 0
回答已采纳
1回答
我试图使用spring-integration-file轮询一个目录，并从这个目录中的文件创建一个反应流。这在很大程度上是可行的，但是当我放置一个文件但没有订阅者就位时，我会得到一个异常。为了演示这个问题，我编写了一个小型演示应用程序：import org.springframework.boot.SpringApplicationorg.springframework.web.bind.annotation.GetMapping;
import org.spri
浏览 9提问于2020-10-31得票数 1
回答已采纳
我有一个用于流的Flux生成器(来自项目反应堆)，它延迟1秒来释放每个元素。由于流是无限的，所以我也决定使用take()方法要使用多少个元素。我想测试这样的方法是否可以在给定的时间内执行。我试图创建我的解决方案，但没有成功。这是Flux方法import java.time.Duration;public class FluxAndMonoStreamsimport org.juni
浏览 2提问于2021-03-12得票数 1
回答已采纳
1回答
是一种源于React团队的单向数据流，它有很多优点(撤消/重做，易于测试，应用程序的一种状态，等等)，将它与AngularJs结合起来会很好。Victor Savkin写了一个关于如何使用AngularJs实现它的，但是他打破了Flux的一个基本概念，改变了存储而不是重新创建它的新实例。另一件让我困扰的事情是，Angular正在使用ng-model和数据绑定，使用Flux意味着我们不应该再使用这些……
考虑到所有这些问题，听起来Angular和Flux范式正在
浏览 0提问于2015-10-29得票数 0
1回答
我正在尝试从server数据库中获取数据(我使用mssql模块来实现：)并将其发送给Redis。所以我创建了三个文件：
sql_server.js包含与sql_server相关的所有函数(例如与数据库和查询的连接)。redis.js具有连接等功能，并将键/值设置为redis数据库(键是ID，值是sql_server查询的结果)。我想对sql_server数据库进行流处理，并将返回的每一行的结果发送给server.js。然后，server.js文件将结果传递
浏览 1提问于2017-09-22得票数 0
回答已采纳
云服务器
ICP备案
云直播
对象存储
实时音视频
腾讯云开发者
