这是Reactor单声道的一个错误吗？

Question

我注意到使用Reactor时的一些奇怪行为。场景是这样的：

• 调用rest APIendpoint，获取一个值，封装在Mono中
• 使用上述值调用另一个rest APIendpoint，检索另一个值，用Mono封装
• 压缩这两个结果

出现的情况是，onSubscribe（FluxMap.MapSubscriber）在第一个API调用中被调用两次，然后打开两个连接并生成两个结果。传递给第二个API调用的结果是不确定的，并且取决于第二个API调用是在前两个调用中的第二个调用完成之前还是之后执行。

这是使用静态编程语言和SpringbootWebClient重现问题的代码示例。APIendpoint要么生成单个GUID，要么根据路径参数生成多个GUID。我使用第一次调用结果中的第一个数字作为第二次调用中的路径参数：

val api = "https://www.uuidgenerator.net/api/guid"
val client = WebClient.builder()
        .baseUrl(api)
        .build()

@Test
public fun reactorBug() {

    val firstResult = callApi().doOnSuccess { r -> println("callApi returned: $r") }
    val secondResult = callApi(firstResult).doOnSuccess { r -> println("callApi(result) returned: $r") }

    println(Mono.zip(firstResult, secondResult, { first, second -> "First result is ${first}Second result is $second" }).block())
}

private fun callApi(): Mono<String> {
    println("Calling Api")
    return client.get().retrieve().bodyToMono()
}

private fun callApi(number: Int): Mono<String> {
    println("Calling Api with $number")
    return client.get().uri("/{number}", number).retrieve().bodyToMono()
}

private fun callApi(firstResult: Mono<String>): Mono<String> {
    println("Extracting number from first result")
    return firstResult
            .map { guid -> guid.find { c -> c.isDigit() } }
            .map { Character.getNumericValue(it!!) }
            .flatMap { i -> callApi(i) }
}

这是说明问题的示例输出：

调用Api从第一个结果中提取编号返回的调用Api:12ec857b-e42c-42ab-a7a2-69beb9a377e3
返回的调用Api:5eedefa5-73b5-4995-aef3-8621e31b698d

编辑的调试输出：

30-01-2018 22:36:11.889[main]DEBUGo.s.web.reactive.function.client.debug-onSubscribe（FLuxMap. MapSubscriber）
30-01-2018 22:36:11.920[main]DEBUGo.s.web.reactive.function.client.debug-请求（无界）
30-01-2018 22:36:11.924[main]DEBUGio.netty.util.NetUtil.debug--Djava.net.preferIPv4Stack： false
30-01-2018 22:36:11.925[main]DEBUGio.netty.util.NetUtil.debug--Djava.net.preferIPv6地址：false
30-01-2018 22:36:12.128[main]DEBUGio.netty.util.NetUtil.debug-环回接口：lo（软件环回接口1，127.0.0.1）
30-01-2018 22:36:12.129[main]DEBUGio.netty.util.NetUtil.debug-无法从sysctl和文件\proc\sys\net\core\Somaxconn获取SOMAXCONN。默认值：200
30-01-2018 22:36:12.146[main]DEBUG r. i. n. r.debugr.i.n.resources.DefaultPoolResources.debug-www.uuidgenerator.net/173.255.225.224的新超文本传输协议客户端池：443
30-01-2018 22:36:12.156[main]DEBUGio.netty.channel.DefaultChannelId.debug-Dio.netty.processId： 4232（自动检测）
30-01-2018 22:36:12.396[main]DEBUGio.netty.channel.DefaultChannelId.debug-Dio.netty.machineId：78:e4:00:ff:fe:bf:a5:cb （自动检测）
30-01-2018 22:36:12.447[main]DEBUGio.netty.buffer.ByteBufUtil.debug-Dio.netty.allocator.type： pooled
30-01-2018 22:36:12.448[main]DEBUGio.netty.buffer.ByteBufUtil.debug-Dio.netty.threadLocalDirectBuffersize： 65536
30-01-2018 22:36:12.448[main]DEBUGio.netty.buffer.ByteBufUmain]DEBUG r. i. n. c. PooledClientContextHandler. debug-从池中获取现有通道：DefaultPromise@d23e4a（不完整）SimpleChannelPool{activeConnections=1}
30-01-2018 22:36:12.461[main]DEBUG o. s. web. reactive. function. client. debug-onSubscribe（FLuxMap. MapSubscript ber）
30-01-2018 22:36:12.462[main]DEBUG o. s. web. reactive. function. client. debug-请求（无界）
30-01-2018 22:36:12.463[main]DEBUG r. i. n. c. PooledClientContextHandler. debug-从池中获取现有通道：DefaultPromise@c8295b（不完整）SimpleChannelPool{activeConnections=1}
30-01-2018 22:36:12.520[reactor-超文本传输协议-nio-2]DefaultPoolResources. debug-创建[id： 0x80971ff0]，现在有2个活动连接

为什么第一个API调用会发生两次？是错误还是Mono的预期行为？

Answer 1

为什么第一次api调用发生两次

zip将订阅firstResult两次，一次直接订阅，一次通过该map-map-194 Map链

在这种情况下，您不需要邮政编码，而只需要来回映射：

val firstResult = callApi().doOnSuccess { r -> println("callApi returned: $r") }

val lastResult = firstResult
        .flatMap { first -> 
            Mono.just(first)
                .map { guid -> guid.find { c -> c.isDigit() } }
                .map { Character.getNumericValue(it!!) }
                .flatMap { i -> callApi(i) }
                .map { second -> "First result is ${first}Second result is $second" }
        }

lastResult.block()