项目反应器-如何从Java 8流创建滑动窗口Flux
项目反应器是一个响应式编程框架，用于构建基于事件流的应用程序。它提供了一种简洁而强大的方式来处理异步和并发编程，特别适用于处理大量的数据流。
在Java 8中，我们可以使用流(Stream)来处理集合数据。而在项目反应器中，我们可以使用Flux来处理事件流。Flux是一个可以发射0到N个元素的发布者(Publisher)，它可以被订阅者(Subscriber)订阅并消费这些元素。
创建滑动窗口的Flux可以通过使用窗口操作符来实现。窗口操作符将事件流分割成多个窗口，每个窗口包含一定数量的元素。滑动窗口是一种特殊类型的窗口，它在每次滑动时都会包含之前窗口中的一部分元素。
在项目反应器中，我们可以使用window操作符来创建滑动窗口的Flux。该操作符接受一个参数，表示每个窗口中的元素数量。例如，我们可以使用以下代码创建一个每3个元素为一组的滑动窗口的Flux：
Flux.range(1, 10)
.window(3)
.flatMap(flux -> flux.collectList())
.subscribe(System.out::println);上述代码中，我们首先使用range方法创建一个包含1到10的整数的Flux。然后，我们使用window操作符将该Flux分割成每3个元素为一组的窗口。接下来，我们使用flatMap操作符将每个窗口中的元素收集到一个列表中。最后，我们订阅该Flux并打印每个窗口的元素列表。
滑动窗口的应用场景包括但不限于数据流分析、实时计算、事件处理等。通过使用滑动窗口，我们可以对连续的事件流进行分组和聚合操作，从而实现更复杂的数据处理逻辑。
腾讯云提供了一系列与项目反应器相关的产品和服务，例如腾讯云函数计算、腾讯云消息队列等。您可以通过访问腾讯云官方网站（https://cloud.tencent.com/）了解更多关于这些产品的详细信息和使用指南。
相关·内容
我有两个对象，每个对象中都包含列表，我的整个spring引导项目都使用项目反应堆结构。我是否应该使用通量内的java流或单映射操作来遍历和操作数据？
Flux.fromIterable(page.getPageFieldData()).subscribeOn(Schedulers.elastic()).map(pageField - > {
clientDataMono.subscribeOn(Schedulers.elastic()).map(clientData - > {
Flux.fromIterable(clientData.get
浏览 0提问于2019-07-25得票数 3
回答已采纳
我已经开始学习Spring web flux了。我目前的需求是在我的项目中使用Spring web flux和AMQP。这个项目是使用rabbitmq发布-订阅AMQP消息传递和项目没有网络dependency.This项目将在纯粹的后端工作。
下面是我的demo服务文件，这个服务在消息负载中获取所需的数据，然后服务将根据负载数据进行处理，并完成需要的任务。
public class HandlerService {
@RabbitListener(queues = "${amqp.queue}")
public void consumeMessageFrom
浏览 18提问于2020-03-03得票数 0
1回答
我使用的是项目反应堆，我想执行以下操作：
@Override
public void run(ApplicationArguments args) {
Flux.from(KafkaReceiver.create(receiverOptions)
.receive()
.map(this::getObject)
.flatMap(this::iterateElasticWrites)
浏览 6提问于2022-04-17得票数 3
回答已采纳
有人能给我解释一下eventbus是如何工作的吗?我看过关于它的所有文档，但我不明白。
我会给你我想要做的，所以我有AccountPresenter和AccountView，所以在我的视图中，我有按钮打开窗口，在其中我创建我的帐户，我想做的是滑动视图和创建的窗口在另一个视图演示者，并可以在AcccountPresenter中使用EventBus的spring vaadin或其他eventBus。我真的不明白这是什么角色。
谢谢
浏览 7提问于2017-02-21得票数 0
Flux.create(sink -> {
sink.error(new IOException());
sink.error(new IOException());
}).subscribe(System.out::println, System.out::println, System.out::println);
我认为(错误地)在第一个错误之后，通量将被处理，并且不再侦听任何信号，但上面的代码创建了一个异常：
java.io.IOException
Exception in thread "main" reactor
浏览 0提问于2018-01-12得票数 1
回答已采纳
Java 8流不允许重用。这就产生了一个难题，即在创建滑动窗口流量来计算x(i)*x(i-1)这样的关系时，如何重用流。
下面的代码基于移位运算符的思想。我使用skip(1)移位第一个流，以创建第二个流。
Flux<Integer> primary = Flux.fromStream(IntStream.range(1, 10).boxed());
Flux<Integer> secondary = primary.skip(1);
primary.zipWith(secondary)
.map(t -> t.getT1() * t.getT2()
浏览 7提问于2017-02-01得票数 6
我开始学习Java的反应式编程。整个反应式范式对我来说都是新的。
在我的学习过程中，我遇到了一些术语/库，比如Spring和RxJava。我希望有人能解释一下区别是什么，你在你的项目中使用了哪一个，以及为什么你做出了这个选择。
我读到Spring实际上在内部使用了WebFlux的Flux/Mono。这将问题简化为Spring Webflux在RxJava上的利弊？
浏览 2提问于2019-06-05得票数 33
回答已采纳
1回答
Zip依赖的通量
、、
简化我拥有的
var flux1 : Flux<Integer>;
var flux2 : Flux<Tuple2<A,B>> = flux1.map( id -> dbFindOperation(id) ).cache();
var flux3 : Flux<A> = flux2.map( tuple -> tuple.getT1() );
var flux4 : Flux<B> = flux2.map( tuple -> tuple.getT2() );
var flux5 : Flux<Tuple3<
浏览 2提问于2020-06-25得票数 0
我使用web flux创建了一个spring boot应用程序。我在控制台中记录了我的结果。我看到我使用Mono创建的方法是在parallel-1线程上执行的，但是我看到了这个actor-tcp-nio-1，Mono的方法在其中执行。这个线程是什么？为什么它不在并行-1线程上执行。 控制台日志 2021-04-04 23:27:04.356 INFO 7225 --- [ parallel-1] reactor.Flux.UsingWhen.1 : onSubscribe(FluxUsingWhen.UsingWhenSubscriber)
2021
浏览 111提问于2021-04-05得票数 0
考虑下面的流量
FluxSink<String> sink;
Flux<String> flux1 = Flux
.<String>create(emitter -> {
sink = emitter;
},...)
.cache()
.publish()
.autoConnect();
因此，要添加/订阅一个项目，我们可以执行sink.next(“4”);
flux1.subscribe(item -> log.info(“item: “+item);
通过过滤flux1，比方说从元素
浏览 2提问于2018-12-07得票数 0
1回答
Flux<Long> flux1 = Flux
.<Long>create(fluxSink -> {
for (long i = 0; i < 20; i++) {
fluxSink.next(i);
}
})
.filter(aLong -> aLong % 2 == 0)
.doOnNext(aLong -> System.out.println("flux 1 : " + aLon
浏览 1提问于2019-03-03得票数 1
回答已采纳
我想在Flux上执行移动窗口计算，并生成一个包含计算值的Flux，但我不知道如何实现这一点。 作为一个简化的例子，假设我有一个整数的Flux，我想用这个Flux中每3个连续整数的和生成一个新的Flux。举例说明： 第一个flux包含从1到8的整数：{1，2，3，4，5，6，7，8} 结果流量应包含和：{1+2+3，2+3+4，3+4+5，4+5+6，5+6+7，6+7+8} 我可以很容易地产生第一个通量，并导出包含连续3个值的通量，如下所示： Flux<Integer> f1 = Flux.range(1,8);
Flux<Flux<Integer>> f
浏览 32提问于2019-06-04得票数 1
回答已采纳
1回答
https://projectreactor.io/docs/core/release/api/reactor/core/publisher/doc-files/marbles/flatMapForFlux.svg 这个问题是关于项目反应堆中Flux publisher中的平面地图功能。根据我们的理解，在平面地图中，我们必须返回一个发布者，而平面地图将在内部订阅该发布者。我们想知道的是线程是如何处理这种情况的。它是多线程还是单线程？
浏览 7提问于2019-06-06得票数 1
1回答
我正在学习反应性流和发布子描述实用程序，并且我正在使用Publisher(在我的例子中是Flux)和订阅者的默认行为。
我有两个场景，它们都有相同数量的Flux元素。但是，当我分析日志时，onSubscribe方法请求的是不同数量的元素(例如，在一种情况下是对无界元素的请求，而在另一种情况下是请求32个元素)。
以下是这两种情况和日志：
System.out.println("*********Calling MapData************");
List<Integer> elements = new ArrayList&l
浏览 8提问于2017-12-29得票数 2
回答已采纳
考虑一个简单的控制器 import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.reactive.function.client.WebClient;
@RestController
public class DtoController {
浏览 54提问于2018-12-19得票数 1
我想使用DirectoryStream与项目反应堆列出目录中的所有文件。
我的尝试是：
Path myDir = Paths.get("C:\\Users\\r.dacanal\\Documents\\Reply\\EDA\\logging-consumer\\input");
DirectoryStream<Path> directoryStream = Files.newDirectoryStream(myDir);
Flux.fromIterable(directoryStream).doOnNext(s -> System.out.println(s)
浏览 5提问于2020-04-17得票数 1
回答已采纳
我正在尝试对Flux进行流式传输，每个元素的延迟为1000毫秒。
@Get(value = "/carts/reactive", produces = MediaType.APPLICATION_JSON_STREAM)
public Flux<Cart> getCartsReactive() {
return Flux.range(1, 10)
.map(integer -> Cart.builder().name("cart" + integer).build())
.delayEl
浏览 2提问于2021-02-24得票数 1
2回答
我的usecase是创建如下所示的反应性端点：
public Flux<ServerEvent> getEventFlux(Long forId){
ServicePoller poller = new ServicePollerImpl();
Map<String,Object> params = new HashMap<>();
params.put("id", forId);
Flux<Long> interval = Flux.interval(Duration.ofMillis(pol
浏览 1提问于2018-06-22得票数 1
回答已采纳
1回答
根据，我期望onErrorContinue将忽略错误元素并继续序列。下面的测试用例异常失败
java.lang.AssertionError:预期"expectNext(12)“失败(预期: onNext(12)；实际: onError(java.lang.RuntimeException：
@Test
public void testOnErrorContinue() throws InterruptedException {
Flux<Integer> fluxFromJust = Flux.just(1, 2,3,4,5)
.c
浏览 3提问于2019-12-29得票数 6
回答已采纳
1回答
对于以下代码的输出，我有一些疑问：
Flux.just("a", "b", "c", "d")
.log(null, Level.INFO, true) // line: 18
.flatMap(value ->
Mono.just(value.toUpperCase()).publishOn(Schedulers.elastic()), 2)
.log(null, Level.INFO, true) // line: 21
.t
浏览 0提问于2018-02-17得票数 1
回答已采纳
1回答
我的一位同事对我说，我不应该处理流，我应该创建一个Flux，而不是使用列表本身，然后处理我的数据到它。
但这对我来说没有意义，因为我认为，反应流通常用于阻塞io，而不是数据处理。
有人能证实我同事提出的新方法是否正确吗？如果它是正确的，它比我以前的方法(性能方面的)有什么优点？
//Model
rootObject{
List<rootNodes> rootNodes
}
//My current code
MonoOfRootObject.map( rootobject.getrootnodes.stream()
.(..do some filtering and repla
浏览 0提问于2019-07-25得票数 0
回答已采纳
2回答
如何使用项目设置流/发布者，该项目可以在一段时间内释放事件，但不需要任何具体的值。我想到了Flux<Void>，但是提到Flux<Void>是一个空流。
例如，我使用一个带有文件系统更改的流( Flux<Path> )。我不需要Path值，而是想添加另一个事件。为了做到这一点，我必须创建虚拟Path对象。
Flux<Path> events = getFilesystemChanges();
events.startWith(Path.of("dummy"))
.map(e -> 42 /* subsequent
浏览 3提问于2020-02-28得票数 1
回答已采纳
假设您有一组Flux，您希望将其与一个双函数一起压缩。
Flux<String> flux1 = Flux.just("A", "B", "C");
Flux<String> flux2 = Flux.just("D", "E", "F");
Flux.zip(flux1, flux2, this::zipString).subscribe(System.out::println);
如果一个对象满足某种约束，下面的这个双函数将返回null。这样我们就可以在压缩Flux之后
浏览 2提问于2017-12-29得票数 5
回答已采纳
我想从一个创建一个。只要StreamObserver不以本地方式实现各个接口(例如，)，就需要这样做。
我想出的大致如下：
final StreamObserver<ProtoResponse>[] streamObserverArray = new StreamObserver[1];
Flux<Response> myFlux Flux.create(sink -> streamObserverArray[0] = new StreamObserver<ProtoResponse>() {
@Override
pu
浏览 4提问于2017-10-23得票数 2
回答已采纳
我有以下示例代码： Flux<Integer> range = Flux.range(0, 10);
Flux<Long> longs = Flux.fromStream(new Random().longs(100, 500).boxed()); // (1)
// Flux<Long> longs = Flux.fromIterable(new Random().longs(100, 500).boxed().limit(30).collect(Collectors.toList())); // (2)
Flux<Tuple2<Integ
浏览 57提问于2019-04-09得票数 1
1回答
Flux.just("a", "b", "c")
.log(null, Level.INFO,true) // line 18
.flatMap(value -> Mono.just(value.toUpperCase())
.publishOn(Schedulers.elastic()), 2)
.log(null, Level.INFO,true) // line 21
.subscribe();
其中一些产出：
浏览 1提问于2018-02-17得票数 2
回答已采纳
1回答
我需要从可分页的REST的所有页面中获取项目。我还需要开始处理项目，一旦它们可用，就不需要等待所有页面被加载。为了做到这一点，我使用Spring及其WebClient，并希望返回Flux<Item>。另外，我使用的REST是有限的，对它的每个响应都包含包含当前限制的详细信息的标题：
当前窗口的大小
当前窗口中的剩余时间
窗口中的请求配额
当前窗口中留下的请求
对单个页面请求的响应如下：
{
"data": [],
"meta": {
"pagination": {
&
浏览 1提问于2018-05-13得票数 5
回答已采纳
我在学习项目反应堆的反应编程。
我有以下测试用例：
@Test
public void createAFlux_just() {
Flux<String> fruitFlux = Flux.just("apple", "orange");
fruitFlux.subscribe(f -> {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace()
浏览 9提问于2020-09-28得票数 2
回答已采纳
我正试着从集合中获得一份文件。文档可能不存在，如果使用null，则返回默认值。
我的查询和转换：
return template.findOne(Query().addCriteria(Criteria.where("id")), DeviceSettings::class.java)
.map {
when (it) {
null -> {
defaultSettings(clock)
浏览 1提问于2018-04-29得票数 0
回答已采纳
了解线程流的简单示例。
gshp订阅INFO 1 INFO reactor.Flux.FlatMap.1 -onSubscribe(FluxFlatMap.FlatMapMain)gshp发布INFO 7INFO reactor.Flux.FlatMap.1 - onNext(6)
在这里，reactor.Flux.FlatMap.1对于gshp订阅了on-1，gshp发布了on-7都很常见。
当我们运行java时，它从主线程开始，之后会发生什么，它是创建gshp订阅or 1还是创建？
@Test
public void setUpTestTest() {
Schedule
浏览 1提问于2020-05-06得票数 1
回答已采纳
1回答
据我所知，下游的其余部分需要在线程池中的线程上处理(我将其设置为1024)
这是我的代码。
Flux<String> ips =
Flux.fromIterable(items).map(Item::getIp);
ips
.publishOn(Schedulers.fromExecutor(Executors.newFixedThreadPool(1024)))
.map(ip -> {
try {
Request request = new Request.Builde
浏览 8提问于2019-07-24得票数 1
回答已采纳
我有一个名为WordListenerHandler的spring-webflux处理程序，我可以从浏览器请求一个HTTP GET。此请求是一个使用Flux：.body(Flux.interval(Duration.ofSeconds(1)).log(), Long.class)的Long流，因此它不会刷新页面而不断更新Flux。我希望显示kafka topi的消息，而不是长值的流。因此，我有一个消费者方法onMessage(ConsumerRecord<String, String> consumerRecord)，它接收消息，并填充一个Queue<String> s
浏览 70提问于2021-03-31得票数 1
回答已采纳
背景:我正在尝试将特定于域的语言的编译器挂接到Maven中。DSL被编译成Java源代码，所以我的想法是编写一个在生成源代码阶段执行的Mojo。该编译器目前是作为Gradle插件实现的，不能处理具有依赖关系的多个项目，在尝试支持项目时，我遇到了Gradle中的问题。我现在正在尝试理解Maven是否能以更好的方式处理这种情况。Gradle的问题与这个问题无关。 我的DSL有像Java这样的包，它们被映射到相同的Java包。对于项目(称为“模块”，如果它们是反应堆构建的一部分，则称为AFAIK )也是如此。在这种情况下，对于每个项目，DSL源代码被编译成Java源代码，以及元数据(每个编译的类都
浏览 16提问于2020-07-25得票数 1
回答已采纳
我想将ArrayList<Apple>转换为Flux<Apple>
Java 8是否为此提供了一个api？
如果没有，我如何将数组列表转换为通量？
浏览 4提问于2021-03-08得票数 4
回答已采纳
1回答
我在Java反应式编程方面还是个新手，我的项目要求我在Java中使用反应式编程实现分页。
例如，如果我遇到一个API，它返回流中的10000条记录，那么我需要返回一个具有适当分页的flux。
有没有人能给我一个好的建议？
这就是我正在遵循的方法。
存储库
public interface CouchBaseRepository extends ReactiveCouchBaseRepository<Book,Integer> {
@Query("#{#n1ql.selectEntity} where name=$1")
public Flux<
浏览 4提问于2018-09-22得票数 5
1回答
对不起，这可能是个愚蠢的问题，但是从文档中还不清楚滑动窗口的测量单位是什么？是毫秒、秒还是流中的项目数？我注意到聚合操作正在产生空的结果，我必须显式地过滤它们，因为可能该窗口没有可用的数据，所以我想最后一点它不是一个选项。
浏览 0提问于2018-07-24得票数 0
回答已采纳
我有一个应用程序，它使用接收来自的事件。我的申请可以总结如下：
@Transactional
@StreamListener(MySink.SINK_NAME)
public void processEvents(Flux<Event> events) {
// Transform events and store them in MongoDB using
// spring-boot-data-mongodb-reactive
...
}
问题是，@Transactional似乎不适用于Spring (至少这是我的印象)，因为如果在编
浏览 2提问于2018-06-11得票数 1
回答已采纳
1回答
在写这个问题的时候，我正在使用SpringFramework5.0.0.M5和Akka 2.4.17。在项目中，我目前正在使用来自Akka的参与者消息和流。我明白了，v5也有流。
根据
作为一个反应性引擎/SPI，反应堆核心和IO模块都为集中的用例公开反应流结构，最终与Spring、RxJava、Akka流、Ratpack相结合.作为一个反应性API，反应堆框架模块将提供丰富的易耗特性，如组合和发布子事件。“
它看起来像Spring在引擎盖下使用Akka流(加上其他东西)。
问题是:从Akka流切换到Spring流可能有哪些优点和缺点？
编辑:这里有更广泛的问题(Reactov对Akka一般)
浏览 2提问于2017-02-25得票数 0
回答已采纳
我正在尝试使用反应器中的SwitchOnFirst运算符，这很神奇--除了有时作为BiFunction的第二个参数传递的转换器似乎不包括第一个元素。基本上，客户端通过RSocket向服务器发送2个项目。代码服务器端如下所示： val socket = new AbstractRSocket() {
override def requestChannel(payloads: Publisher[Payload]): Flux[Payload] =
Flux.from(payloads).log.switchOnFirst((signal, all) => handl
浏览 47提问于2020-02-02得票数 0
回答已采纳
下面的多线程示例从两个不同的线程写入TopicProcessor，并在两个不同的线程中从TopicProcessor读取。然而，在某些地方存在竞争条件，使得不是所有事件都被传递给订阅者，从而导致应用程序在processed.await()中永远挂起。有人知道为什么吗？
import reactor.core.publisher.Flux;
import reactor.extra.processor.TopicProcessor;
import reactor.extra.processor.WaitStrategy;
import java.time.Duration;
import ja
浏览 0提问于2020-02-06得票数 2
1回答
我想要实时更新我使用webflux创建的流。以下是我的代码：
public Flux<UserEvent> getEvents(String userId)
{
return findById(userId)
.flatMapMany(user -> {
Flux<Long> interval = Flux.interval(Duration.ofSeconds(2));
Flux<UserEvent> userEve
浏览 7提问于2018-07-25得票数 0
在ProjectReactor或反应性流中，在订阅().之前，不会发生任何事情。
除非有人订阅了，否则将不会发生反应性流数据流，但我看到对于所有REST API，如find、save和inserts并不是显式调用订阅，而是在生产者和订阅者之间流动数据。
@RestController
class PersonController {
private final PersonRepository repository;
public PersonController(PersonRepository repository) {
this.reposit
浏览 1提问于2018-06-11得票数 10
回答已采纳
我有简单的微控制器与两个类似的GET方法，返回不同的反应类型。
@Controller
public class MyController {
@Get(uri = "/a", produces = MediaType.APPLICATION_JSON_STREAM)
public Flux<MyPojo> getA() {
return Flux.just("Hi")
.flatMap(s -> Mono.error(new HttpStatusException(Ht
浏览 8提问于2022-01-11得票数 1
使用spring反应性应用程序，我创建了一个rest服务，它每秒钟产生一个事件。rest控制器的代码是：
@GetMapping(value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<Event> getEvents() {
Flux<Event> eventFlux = Flux.fromStream(Stream.generate(() -> new Event(new Random().nextLong(), "Hello Ev
浏览 2提问于2018-04-22得票数 1
回答已采纳
我必须用window size = 3创建一个输入数据的滑动窗口
Dataframe
0 1
0 1 2
1 3 4
2 5 6
3 7 8
期望输出：
0 1 2 3 4 5
0 1 2 NA NA NA NA
1 3 4 1 2 NA NA
2 5 6 3 4 1 2
3 7 8 5 6 3 4
我使用了data.values.flatten()，但它在dataframe中转换nested list中的所有行。
浏览 2提问于2020-09-11得票数 1
回答已采纳
我正在使用Kotlin开发一个Spring (2.2)项目，使用CouchDB作为(反应性)数据库，因此使用异步DAO (挂起函数或返回流的函数)。我正在尝试设置WebFlux，以便拥有异步控制器(同样，我希望返回流，而不是Flux)。但我在从ReactiveSecurityContextHolder检索安全上下文时遇到了麻烦。
据我所读，与使用SecurityContextHolder存储它的ThreadLocal不同，ReactiveSecurityContextHolder依赖于这样一个事实，即Spring在订阅我的反应性链时，也将该上下文存储在这个链中，从而允许我从链中调用Reacti
浏览 2提问于2019-11-26得票数 7
回答已采纳
我尝试在我的TYPO3后端(v6.2)用滑动内容元素。所以我不包括CSS_Styled_content。示例：
10 < styles.content.getRight
10.slide = -1
特别是我想将我的内容从colPos=0 'Slider'幻灯片到下面的页面。这是可能的流体/ FluidTYPO3 (流量)吗？
<f:section name="Configuration">
<flux:form id="fluidpage" options="{icon: 'typo3conf
浏览 6提问于2015-02-13得票数 0
回答已采纳
1回答
我运行了以下代码：
Mono<Void> mono = Mono.empty();
System.out.println("mono.block: " + mono.block());
它会产生：
mono.block: null
不出所料。换句话说，如果Mono已经完成，调用block将立即返回。
另一个例子，类似于真实世界的场景。我有一个源流，例如：
Flux<Integer> ints = Flux.range(0, 2);
我创建了一个可连接的flux，我将使用它来允许多个订阅者：
ConnectableFlux<Integer&
浏览 1提问于2019-04-13得票数 0
您好，我的代码如下所示： fun mapBatch(batch: List<String>): Mono<List<MyClass>> ...
fun myFun(stream: Flux<String>): Flux<MyClass> {
return stream
.bufferTimeout(50, Duration.ofSeconds(60L))
.flatMap{ batch -> mapBatch(batch) }
/// now h
浏览 132提问于2021-06-14得票数 2
回答已采纳
我试图用从Mono接收的数据来操作从Flux接收的对象，其中发出对象的流量和项的Mono的方法都是不同的API调用。问题是，我无法控制线程，从Mono接收的项从未分配给我的对象，除非我有意阻止该线程。请建议在这种情况下是否可以采用任何非阻塞的方式。
我还研究过调度程序、subscribeOn、publishOn，但无法找到管道。
public Flux<Object> test {
method1().map(obj -> {
if (obj.getTotalItems() > 20) {
obj.setItems(metho
浏览 0提问于2019-01-17得票数 1
回答已采纳
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM
腾讯云开发者
