在我的生产代码中，当 Mono 超时时，我的日志中出现错误。
我已设法使用以下代码重新创建这些错误：

@Test
public void testScheduler() {
    Mono<String> callableMethod1 = callableMethod();
    callableMethod1.block();

    Mono<String> callableMethod2 = callableMethod();
    callableMethod2.block();
}

private Mono<String> callableMethod() {
    return Mono.fromCallable(() -> {
        Thread.sleep(60);
        return "Success";
    })
            .subscribeOn(Schedulers.elastic())
            .timeout(Duration.ofMillis(50))
            .onErrorResume(throwable -> Mono.just("Timeout"));
}

在Mono.fromCallable我正在使用第三方库进行阻塞调用。当此调用超时时，我收到类似于

reactor.core.publisher.Operators - Operator called default onErrorDropped
reactor.core.publisher.Operators - Scheduler worker in group main failed with an uncaught exception

这些错误似乎也是间歇性的，有时当我运行代码时，我根本没有得到任何错误。但是，当我以 10 的循环重复调用时，我始终得到它们。

Sinks.Many<String>在使用向多个订阅者通知某些事件时，我遇到了一种我不明白的行为：

fun main() {

    val sink : Sinks.Many<String>  = Sinks.many().multicast().onBackpressureBuffer()
    val flux = sink.asFlux().log()

    val d = flux.subscribe {
        println("--> $it")
    }

    sink.emitNext("1", Sinks.EmitFailureHandler.FAIL_FAST)

    val d2 = flux.subscribe {
        println("--2> $it")
    }

    sink.emitNext("2", Sinks.EmitFailureHandler.FAIL_FAST)
}

此代码显示第一个订阅者获取值 1 和 2，第二个订阅者获取值 2。到目前为止一切顺利：

11:49:06.936 [main] INFO reactor.Flux.EmitterProcessor.1 - onSubscribe(EmitterProcessor.EmitterInner)
11:49:06.938 [main] INFO reactor.Flux.EmitterProcessor.1 - request(unbounded)
11:49:06.942 [main] INFO reactor.Flux.EmitterProcessor.1 - onNext(1)
--> 1
11:49:06.942 [main] INFO reactor.Flux.EmitterProcessor.1 - onSubscribe(EmitterProcessor.EmitterInner)
11:49:06.942 [main] INFO reactor.Flux.EmitterProcessor.1 - request(unbounded)
11:49:06.943 [main] INFO reactor.Flux.EmitterProcessor.1 - onNext(2)
--> …

我正在将用RxJava 1.x编写的小项目转换为Reactor 3.x. 一切都很好,除了我无法找到如何替换flatMap(Observable::from)适当的对应物.我有Mono<List<String>>,我需要将其转换为Flux<String>.

谢谢

Mono.delay(Duration.ofMillis(10)).map(d -> {
            System.out.println(d);
            return d;
        }).block();

输出：0

当我使用subscribe()或subscribe(-)方法而不是调用时，无法在控制台上看到任何输出block()

Mono.delay(Duration.ofMillis(10)).map(d -> {
        System.out.println(d);
        return d;
    }).subscribe(System.out::println);

doOnSubscribe(-)难道我们只需要在这个方法之后使用吗Mono.delay(-)？

 Mono.delay(Duration.ofMillis(10)).doOnSubscribe(s -> {
        System.out.println("its printing doOnSubscribe");
    }).map(d -> {
        System.out.println(d);
        return d;
    }).subscribe(System.out::println);

输出它正在打印doOnSubscribe

我正在使用WebFlux框架开发一个使用Spring Boot 2.0和Kotlin的应用程序.

我想在保存交易之前检查用户ID是否退出.我被困在一个简单的事情,如验证Mono是否为空.

fun createTransaction(serverRequest: ServerRequest) : Mono<ServerResponse> {
    val transaction = serverRequest.body(BodyExtractors.toMono(Transaction::class.java))

    transaction.flatMap {
        val user = userRepository.findById(it.userId)
        // If it's empty, return badRequest() 
    } 

    return transaction.flatMap { transactionRepository.save(it).then(created(URI.create("/transaction/" + it.id)).build()) }
}

有可能做我想要的吗？

允许对结果流进行多次迭代,CompletableFuture<Stream<String>>我正在考虑以下方法之一:

1. 将生成的未来转换为CompletableFuture<List<String>>:teams.thenApply(st -> st.collect(toList()))

2. 将生成的未来转换为Flux<String>缓存:Flux.fromStream(teams::join).cache();

Flux<T>是Publisher<T>项目反应堆的实施.

使用案例:

我想Stream<String>从一个数据源获得一个具有顶级联赛球队名称的序列(例如),该数据源提供一个League对象Standing[](基于足球数据RESTful API,例如http://api.football-data.org/v1/ soccerseasons/445/leagueTable).使用AsyncHttpClient和Gson我们有:

CompletableFuture<Stream<String>> teams = asyncHttpClient
    .prepareGet("http://api.football-data.org/v1/soccerseasons/445/leagueTable")
    .execute()
    .toCompletableFuture()
    .thenApply(Response::getResponseBody)
    .thenApply(body -> gson.fromJson(body, League.class));
    .thenApply(l -> stream(l.standings).map(s -> s.teamName));

要重新使用生成的流,我有两个选择:

1. CompletableFuture<List<String>> res = teams.thenApply(st -> st.collect(toList()))

2. Flux<String> res = Flux.fromStream(teams::join).cache()

Flux<T>不那么冗长,并提供我所需要的一切.然而,在这种情况下使用它是否正确？

或者我应该使用CompletableFuture<List<String>>？或者还有其他更好的选择吗？

更新了一些想法(2018-03-16):

CompletableFuture<List<String>>:

• [PROS] List<String>将继续收集,当我们需要继续处理未来的结果时,可能已经完成了.
• [CONS]宣言冗长.
• [CONS]如果我们只想使用它一次,那么我们就不需要收集那些物品了List<T>.

Flux<String> …