这是反应堆Mono里的窃听器吗？

Question

我在使用反应堆时注意到了一些奇怪的行为。设想的情况是：

• 调用rest端点，获取一个值，包装在Mono中
• 用上面的值调用另一个rest端点，检索另一个值，包装在Mono中
• 压缩这两个结果

对于第一个API调用，onSubscribe(FluxMap.MapSubscriber)会被调用两次，然后打开两个连接并产生两个结果。传递给第二个API调用的结果是不确定的，并取决于第二个API调用是在前两个调用结束之前还是之后执行。

这是使用Kotlin和Springboot WebClient再现问题的代码示例。API端点要么生成单个GUID，要么根据path参数生成许多GUID。我在第一次调用的结果中使用第一个数字作为第二次调用中的路径param：

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 从第一个结果中提取数字 callApi返回: 12ec857b-e42c-42ab-a7a2-69beb9a377e3 callApi返回:5eedefa5-73b5-4995-aef3-8621e31b698 d <-此结果不应发生 使用5<调用Api --这应该是1，而不是5 返回的callApi(结果)： 01c64488-6a8c-4400-9094-6729c64a4e1a 0179 40b6 d2b4-40b6-8489-52 fa58de25f 8f814b1d-594 c-4392-a4f5-04d417367 45891d71-61b2-4d5b-81 4d5b 2 cfd8e453377 08edf0c3-3614-402 b-8b17-000 fce1a0 第一个结果是12ec857b-e42c-42ab-a7a2-69beb9a377e3。 第二个结果是 01c64488-6a8c-4400-9094-6729c64a4e1a 0179 40b6 d2b4-40b6-8489-52 fa58de25f 8f814b1d-594 c-4392-a4f5-04d417367 45891d71-61b2-4d5b-81 4d5b 2 cfd8e453377 08edf0c3-3614-402 b-8b17-000 fce1a0

编辑的调试输出：

onSubscribe(FluxMap.MapSubscriber) 30-01-2018 22:36:11.889主要调试o.s.web.reactive.function.client.debug -o.s.web.reactive.function.client.debug

30-01-2018 22:36:11.920主调试o.s.web.reactive.function.client.debug -请求(无限)

2018年22:36:11.924主要调试io.netty.util.NetUtil.debug - -Djava.net.preferIPv4Stack: false

2018年22:36:11.925主要调试io.netty.util.NetUtil.debug - -Djava.net.preferIPv6Addresses: false

30-01-2018 22:36:12.128主要调试io.netty.util.NetUtil.debug -回溯接口: lo (软件回溯接口1,127.0.0.1)

30-01-2018 22:36:12.129主要调试io.netty.util.NetUtil.debug -从sysctl和文件\proc\sys\net\core\somaxconn获得SOMAXCONN失败。违约: 200

30-01-2018 22:36:12.146主调试r.i.n.r.DefaultLoopEpollDetector.debug默认支持: false

2018-22:36:12.156主要调试r.i.n.resources.DefaultPoolResources.debug -www.uidgenerator.net/173.255.225.224:443的新http客户机池

2018年22:36:12.190主要调试io.netty.channel.DefaultChannelId.debug - -Dio.netty.processId: 4232 (自动检测)

主要调试io.netty.channel.DefaultChannelId.debug：-Dio.netty.machineId: 78:e4:00:ff:fe:bf:a5:cb (自动检测)

30-01-2018 22:36:12.447主调试io.netty.buffer.ByteBufUtil.debug - -Dio.netty.allocator.type:池

2018年22:36:12.448主要调试io.netty.buffer.ByteBufUtil.debug - -Dio.netty.threadLocalDirectBufferSize: 65536

2018年22:36:12.448主要调试io.netty.buffer.ByteBufUtil.debug - -Dio.netty.maxThreadLocalCharBufferSize: 16384

30-01-2018 22:36:12.459主调试r.i.n.c.PooledClientContextHandler.debug -从池获取现有信道:DefaultPromise@d23e4a(不完全) SimpleChannelPool{activeConnections=1}

onSubscribe(FluxMap.MapSubscriber) 30-01-2018 22:36:12.461主要调试o.s.web.reactive.function.client.debug -o.s.web.reactive.function.client.debug

30-01-2018 22:36:12.462主调试o.s.web.reactive.function.client.debug -请求(无限)

30-01-2018 22:36:12.463主调试r.i.n.c.PooledClientContextHandler.debug -从池获取现有信道:DefaultPromise@c8295b(不完全) SimpleChannelPool{activeConnections=1}

30-01-2018 22:36:12.520反应堆-http-nio-2除错r.i.n.resources.DefaultPoolResources.debug -创建id: 0x88225196，现在2个活动连接

30-01-2018 22:36:12.520反应堆-http-nio-4调试r.i.n.resources.DefaultPoolResources.debug -创建id: 0x80971ff0，现在有2个活动连接

为什么第一个API调用会发生两次-它是一个bug还是这是Mono的预期行为？

Answer 1

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

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

在这种情况下，您不需要zip，而只需要来回执行flatMap：

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()