Spring WebFlux：只允许一个连接接收订阅者
Spring WebFlux: Only one connection receive subscriber allowed
我正在使用 Spring 5 Webflux 和 Kotlin 编写一个简单的应用程序。我正在尝试通过以下方式实现 PUT 端点：
PUT("/confs/{id}", {
val id = it.pathVariable("id")
ServerResponse.ok().body(service.save(it.bodyToMono(Item::class.java)), Item::class.java)
})
保存的技巧是我尝试从项目中读取城市名称，解析地理坐标，在原始项目中覆盖它们，然后使用 Spring 数据 Mongo 保存到 Mongo ] 反应性回购。
fun save(item: Mono<Item>): Mono<Item> {
val geo = item.flatMap {
val city = it.location?.city ?: "Somewhere"
geoService.resolveGeoFromCity(city)
}
val zipped = item.zipWith(geo)
.map {
it.t1.location?.geo = it.t2
it.t1
}
return repo.saveAll(zipped)
.toMono()
}
解析地理坐标的代码在这里：
@Service
class GeoService() {
val client = WebClient.create("https://maps.googleapis.com/maps/api/geocode/")
fun resolveGeoFromCity(city: String): Mono<Geo> {
return client.get()
.uri("json?address=$city&key=$API_KEY&language=en")
.exchange()
.flatMap { it.bodyToMono(String::class.java) }
.map { parse(it) }
}
private fun parse(response: String): Geo {
val locationMap = JsonPath.read<Map<String, Double>>(response, "$.results[0].geometry.location")
return Geo(locationMap["lat"] ?: 0.0, locationMap["lng"] ?: 0.0)
}
}
问题是，如果发出 PUT 请求，我会得到跟踪堆栈跟踪。我尝试用 val geo = Mono.just(Geo(0.0, 0.0)) 存根 Mono（不使用 WebClient）然后它工作正常。
如何在不牺牲功能的情况下修复它？）
2018-01-01 01:41:00.595 ERROR 15120 --- [ctor-http-nio-4] .a.w.r.e.DefaultErrorWebExceptionHandler : Failed to handle request [PUT http://localhost:8097/confs/5a49675c910d123b1057207a]
java.lang.IllegalStateException: Only one connection receive subscriber allowed.
at reactor.ipc.netty.channel.FluxReceive.startReceiver(FluxReceive.java:276) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE]
at reactor.ipc.netty.channel.FluxReceive.subscribe(FluxReceive.java:124) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE]
at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.ipc.netty.ByteBufFlux.subscribe(ByteBufFlux.java:242) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE]
at reactor.core.publisher.FluxPeek.subscribe(FluxPeek.java:83) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.ipc.netty.ByteBufFlux.subscribe(ByteBufFlux.java:242) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE]
at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxFlatMap.subscribe(FluxFlatMap.java:97) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxDoFinallyFuseable.subscribe(FluxDoFinallyFuseable.java:48) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~
[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:74) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.Mono.subscribe(Mono.java:3008) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:167) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:71) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.ipc.netty.channel.ChannelOperations.applyHandler(ChannelOperations.java:383) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE]
at reactor.ipc.netty.http.server.HttpServerOperations.onHandlerStart(HttpServerOperations.java:359) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE]
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:163) ~[netty-common-4.1.17.Final.jar:4.1.17.Final]
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java) ~[netty-common-4.1.17.Final.jar:4.1.17.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:403) ~[netty-common-4.1.17.Final.jar:4.1.17.Final]
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:463) ~[netty-transport-4.1.17.Final.jar:4.1.17.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor.run(SingleThreadEventExecutor.java:858) ~[netty-common-4.1.17.Final.jar:4.1.17.Final]
at java.lang.Thread.run(Thread.java:745) ~[na:1.8.0_112]
Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Assembly trace from producer [reactor.core.publisher.FluxMap] :
reactor.core.publisher.Flux.map(Flux.java:4855)
reactor.ipc.netty.ByteBufFlux.fromInbound(ByteBufFlux.java:68)
reactor.ipc.netty.NettyInbound.receive(NettyInbound.java:90)
org.springframework.http.server.reactive.ReactorServerHttpRequest.getBody(ReactorServerHttpRequest.java:148)
org.springframework.http.codec.DecoderHttpMessageReader.readMono(DecoderHttpMessageReader.java:93)
org.springframework.http.codec.DecoderHttpMessageReader.readMono(DecoderHttpMessageReader.java:123)
org.springframework.web.reactive.function.BodyExtractors.lambda$null[=14=](BodyExtractors.java:101)
java.util.Optional.map(Optional.java:215)
org.springframework.web.reactive.function.BodyExtractors.readWithMessageReaders(BodyExtractors.java:256)
org.springframework.web.reactive.function.BodyExtractors.lambda$toMono(BodyExtractors.java:96)
org.springframework.web.reactive.function.server.DefaultServerRequest.body(DefaultServerRequest.java:126)
org.springframework.web.reactive.function.server.DefaultServerRequest.body(DefaultServerRequest.java:120)
org.springframework.web.reactive.function.server.DefaultServerRequest.bodyToMono(DefaultServerRequest.java:145)
com.example.confs.web.ConferenceRouter$routes.invoke(ConferenceRouter.kt:31)
com.example.confs.web.ConferenceRouter$routes.invoke(ConferenceRouter.kt:16)
org.springframework.web.reactive.function.server.RouterFunctionDsl$PUT.handle(RouterFunctionDsl.kt:200)
org.springframework.web.reactive.function.server.support.HandlerFunctionAdapter.handle(HandlerFunctionAdapter.java:61)
org.springframework.web.reactive.DispatcherHandler.invokeHandler(DispatcherHandler.java:168)
org.springframework.web.reactive.DispatcherHandler.lambda$handle(DispatcherHandler.java:160)
reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:118)
reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67)
reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:76)
reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerNext(FluxConcatMap.java:271)
reactor.core.publisher.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:803)
reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115)
reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67)
reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67)
reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67)
reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67)
reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:1649)
reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:1463)
reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:1337)
reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
reactor.core.publisher.Mono.subscribe(Mono.java:3008)
reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onComplete(FluxSwitchIfEmpty.java:75)
reactor.core.publisher.Operators.complete(Operators.java:125)
reactor.core.publisher.MonoEmpty.subscribe(MonoEmpty.java:45)
reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
reactor.core.publisher.Mono.subscribe(Mono.java:3008)
reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:418)
reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onSubscribe(FluxConcatMap.java:210)
reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:128)
reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:61)
reactor.core.publisher.FluxConcatMap.subscribe(FluxConcatMap.java:121)
reactor.core.publisher.MonoNext.subscribe(MonoNext.java:40)
reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60)
reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60)
reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61)
reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:74)
reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
reactor.core.publisher.Mono.subscribe(Mono.java:3008)
reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:167)
reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56)
reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61)
reactor.ipc.netty.channel.ChannelOperations.applyHandler(ChannelOperations.java:383)
reactor.ipc.netty.http.server.HttpServerOperations.onHandlerStart(HttpServerOperations.java:359)
io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:163)
io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:403)
io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:463)
io.netty.util.concurrent.SingleThreadEventExecutor.run(SingleThreadEventExecutor.java:858)
Error has been observed by the following operator(s):
|_ Flux.map(ByteBufFlux.java:68)
|_ Flux.doOnNext(ByteBufFlux.java:230)
|_ Flux.map(ReactorServerHttpRequest.java:148)
|_ Flux.flatMap(AbstractJackson2Decoder.java:95)
|_ Flux.doFinally(AbstractJackson2Decoder.java:95)
|_ Flux.map(AbstractJackson2Decoder.java:117)
|_ Flux.singleOrEmpty(AbstractJackson2Decoder.java:87)
|_ Operators.error(FluxReceive.java:276)
|_ Mono.onErrorMap(DefaultServerRequest.java:146)
|_ Mono.map(ConferenceService.kt:27)
|_ Mono.map(ConferenceService.kt:32)
|_ Mono.zipWith(ConferenceService.kt:47)
|_ Mono.map(ConferenceService.kt:48)
|_ Flux.flatMap(SimpleReactiveMongoRepository.java:318)
|_ MonoExtensionsKt.toMono(ConferenceService.kt:55)
|_ Mono.map(ConferenceService.kt:56)
|_ Flux.map(AbstractJackson2Encoder.java:99)
|_ Mono.flatMap(DispatcherHandler.java:177)
|_ Mono.onErrorResume(DispatcherHandler.java:177)
|_ Mono.flatMap(DispatcherHandler.java:161)
|_ Mono.defer(DefaultWebFilterChain.java:71)
|_ Mono.doOnSuccess(MetricsWebFilter.java:59)
|_ Mono.doOnError(MetricsWebFilter.java:60)
|_ Mono.compose(MetricsWebFilter.java:54)
|_ Mono.defer(DefaultWebFilterChain.java:71)
|_ Mono.defer(DefaultWebFilterChain.java:71)
我在这里做了一个非常相似的例子：
此路由器将从名称中获取地理位置，并使用另一项服务提取日出和日落时间：
我使用了 Mono 的 and 方法。
internal fun buildResponse(address: Mono<String>) =
address.transform(geoLocationService::fromAddress).and(this::sunriseSunset, ::LocationResponse)
internal fun sunriseSunset(geographicCoordinates: GeographicCoordinates) =
geographicCoordinates.toMono().transform(sunriseSunsetService::fromGeographicCoordinates)
更多详情example
WebClient.exchange() 结果流是单播
这里的问题实际上是 WebClient 每个连接只允许一个订阅者。如果您尝试订阅同一个 exchanged 连接两次 - 您将获得 java.lang.IllegalStateException: Only one connection receive subscriber allowed.
尽管我没有看到您在何处尝试重复使用同一连接两次，但我相信您可以通过使用下一个运算符组合来解决该问题：
class GeoService() {
val client = WebClient.create("https://maps.googleapis.com/maps/api/geocode/")
fun resolveGeoFromCity(city: String): Mono<Geo> {
return client.get()
.uri("json?address=$city&key=$API_KEY&language=en")
.exchange()
.flatMap { it.bodyToMono(String::class.java) }
.map { parse(it) }
.share();
}
...
}
在该示例中，流配置为多播（共享）原始源，只要至少有一个 Subscriber 将被订阅。
如果您需要所有订阅者收到相同的日期，您可以将 .share 替换为 .cache 运算符。
此外，还有一种方法可以替代上述技术。您可以用处理器替换提到的运算符并获得相同的共享可能性：
class GeoService() {
val client = WebClient.create("https://maps.googleapis.com/maps/api/geocode/")
fun resolveGeoFromCity(city: String): Mono<Geo> {
return client.get()
.uri("json?address=$city&key=$API_KEY&language=en")
.exchange()
.flatMap { it.bodyToMono(String::class.java) }
.map { parse(it) }
.subscribeWith(DirectProcessor.create());
}
...
}
在那种情况下，您在调用 subscribeWith 之后就订阅并 运行 使用源的数据，因此，在这种情况下，您可能会丢失部分数据，等等。
为什么 Mono.just(..) 一切正常？
首先.just是一个冷运算符，它允许尽可能多的订阅者在任何时间点接收相同的数据。这就是为什么当您尝试两次从连接中使用相同的数据块时，您没有得到任何异常。
我遇到了类似的问题。解决方法是指定此依赖项：
org.springframework:spring-webflux:5.1.4.RELEASE
当我使用 spring-boot 时，它会部署它的先前版本。不幸的是，link 现在不存在此问题。
所以现在我的 gradle 看起来像这样：
compile('org.springframework.boot:spring-boot-starter-data-mongodb-reactive')
compile('org.springframework.boot:spring-boot-starter-webflux')
// Next 2 dependencies are temporally here until the one above does not resolves next to at least 5.1.4 - where
// webflux issue is resolved:
// https://github.com/rstoyanchev/spr-issue-migration-test-2/issues/17323
compile('org.springframework:spring-webflux:5.1.4.RELEASE')
compile('org.springframework:spring-web:5.1.4.RELEASE')
我正在使用 Spring 5 Webflux 和 Kotlin 编写一个简单的应用程序。我正在尝试通过以下方式实现 PUT 端点：
PUT("/confs/{id}", {
val id = it.pathVariable("id")
ServerResponse.ok().body(service.save(it.bodyToMono(Item::class.java)), Item::class.java)
})
保存的技巧是我尝试从项目中读取城市名称，解析地理坐标，在原始项目中覆盖它们，然后使用 Spring 数据 Mongo 保存到 Mongo ] 反应性回购。
fun save(item: Mono<Item>): Mono<Item> {
val geo = item.flatMap {
val city = it.location?.city ?: "Somewhere"
geoService.resolveGeoFromCity(city)
}
val zipped = item.zipWith(geo)
.map {
it.t1.location?.geo = it.t2
it.t1
}
return repo.saveAll(zipped)
.toMono()
}
解析地理坐标的代码在这里：
@Service
class GeoService() {
val client = WebClient.create("https://maps.googleapis.com/maps/api/geocode/")
fun resolveGeoFromCity(city: String): Mono<Geo> {
return client.get()
.uri("json?address=$city&key=$API_KEY&language=en")
.exchange()
.flatMap { it.bodyToMono(String::class.java) }
.map { parse(it) }
}
private fun parse(response: String): Geo {
val locationMap = JsonPath.read<Map<String, Double>>(response, "$.results[0].geometry.location")
return Geo(locationMap["lat"] ?: 0.0, locationMap["lng"] ?: 0.0)
}
}
问题是，如果发出 PUT 请求，我会得到跟踪堆栈跟踪。我尝试用 val geo = Mono.just(Geo(0.0, 0.0)) 存根 Mono（不使用 WebClient）然后它工作正常。
如何在不牺牲功能的情况下修复它？）
2018-01-01 01:41:00.595 ERROR 15120 --- [ctor-http-nio-4] .a.w.r.e.DefaultErrorWebExceptionHandler : Failed to handle request [PUT http://localhost:8097/confs/5a49675c910d123b1057207a]
java.lang.IllegalStateException: Only one connection receive subscriber allowed.
at reactor.ipc.netty.channel.FluxReceive.startReceiver(FluxReceive.java:276) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE]
at reactor.ipc.netty.channel.FluxReceive.subscribe(FluxReceive.java:124) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE]
at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.ipc.netty.ByteBufFlux.subscribe(ByteBufFlux.java:242) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE]
at reactor.core.publisher.FluxPeek.subscribe(FluxPeek.java:83) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.ipc.netty.ByteBufFlux.subscribe(ByteBufFlux.java:242) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE]
at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxFlatMap.subscribe(FluxFlatMap.java:97) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxDoFinallyFuseable.subscribe(FluxDoFinallyFuseable.java:48) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.FluxOnAssembly.subscribe(FluxOnAssembly.java:252) ~
[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:74) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.Mono.subscribe(Mono.java:3008) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:167) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:71) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:76) ~[reactor-core-3.1.2.RELEASE.jar:3.1.2.RELEASE]
at reactor.ipc.netty.channel.ChannelOperations.applyHandler(ChannelOperations.java:383) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE]
at reactor.ipc.netty.http.server.HttpServerOperations.onHandlerStart(HttpServerOperations.java:359) ~[reactor-netty-0.7.2.RELEASE.jar:0.7.2.RELEASE]
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:163) ~[netty-common-4.1.17.Final.jar:4.1.17.Final]
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java) ~[netty-common-4.1.17.Final.jar:4.1.17.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:403) ~[netty-common-4.1.17.Final.jar:4.1.17.Final]
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:463) ~[netty-transport-4.1.17.Final.jar:4.1.17.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor.run(SingleThreadEventExecutor.java:858) ~[netty-common-4.1.17.Final.jar:4.1.17.Final]
at java.lang.Thread.run(Thread.java:745) ~[na:1.8.0_112]
Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Assembly trace from producer [reactor.core.publisher.FluxMap] :
reactor.core.publisher.Flux.map(Flux.java:4855)
reactor.ipc.netty.ByteBufFlux.fromInbound(ByteBufFlux.java:68)
reactor.ipc.netty.NettyInbound.receive(NettyInbound.java:90)
org.springframework.http.server.reactive.ReactorServerHttpRequest.getBody(ReactorServerHttpRequest.java:148)
org.springframework.http.codec.DecoderHttpMessageReader.readMono(DecoderHttpMessageReader.java:93)
org.springframework.http.codec.DecoderHttpMessageReader.readMono(DecoderHttpMessageReader.java:123)
org.springframework.web.reactive.function.BodyExtractors.lambda$null[=14=](BodyExtractors.java:101)
java.util.Optional.map(Optional.java:215)
org.springframework.web.reactive.function.BodyExtractors.readWithMessageReaders(BodyExtractors.java:256)
org.springframework.web.reactive.function.BodyExtractors.lambda$toMono(BodyExtractors.java:96)
org.springframework.web.reactive.function.server.DefaultServerRequest.body(DefaultServerRequest.java:126)
org.springframework.web.reactive.function.server.DefaultServerRequest.body(DefaultServerRequest.java:120)
org.springframework.web.reactive.function.server.DefaultServerRequest.bodyToMono(DefaultServerRequest.java:145)
com.example.confs.web.ConferenceRouter$routes.invoke(ConferenceRouter.kt:31)
com.example.confs.web.ConferenceRouter$routes.invoke(ConferenceRouter.kt:16)
org.springframework.web.reactive.function.server.RouterFunctionDsl$PUT.handle(RouterFunctionDsl.kt:200)
org.springframework.web.reactive.function.server.support.HandlerFunctionAdapter.handle(HandlerFunctionAdapter.java:61)
org.springframework.web.reactive.DispatcherHandler.invokeHandler(DispatcherHandler.java:168)
org.springframework.web.reactive.DispatcherHandler.lambda$handle(DispatcherHandler.java:160)
reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:118)
reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67)
reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:76)
reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerNext(FluxConcatMap.java:271)
reactor.core.publisher.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:803)
reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115)
reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67)
reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67)
reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67)
reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67)
reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:1649)
reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:1463)
reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:1337)
reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
reactor.core.publisher.Mono.subscribe(Mono.java:3008)
reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onComplete(FluxSwitchIfEmpty.java:75)
reactor.core.publisher.Operators.complete(Operators.java:125)
reactor.core.publisher.MonoEmpty.subscribe(MonoEmpty.java:45)
reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
reactor.core.publisher.Mono.subscribe(Mono.java:3008)
reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:418)
reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onSubscribe(FluxConcatMap.java:210)
reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:128)
reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:61)
reactor.core.publisher.FluxConcatMap.subscribe(FluxConcatMap.java:121)
reactor.core.publisher.MonoNext.subscribe(MonoNext.java:40)
reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60)
reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60)
reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61)
reactor.core.publisher.MonoPeekFuseable.subscribe(MonoPeekFuseable.java:74)
reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
reactor.core.publisher.Mono.subscribe(Mono.java:3008)
reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:167)
reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56)
reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
reactor.core.publisher.MonoPeekTerminal.subscribe(MonoPeekTerminal.java:61)
reactor.ipc.netty.channel.ChannelOperations.applyHandler(ChannelOperations.java:383)
reactor.ipc.netty.http.server.HttpServerOperations.onHandlerStart(HttpServerOperations.java:359)
io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:163)
io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:403)
io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:463)
io.netty.util.concurrent.SingleThreadEventExecutor.run(SingleThreadEventExecutor.java:858)
Error has been observed by the following operator(s):
|_ Flux.map(ByteBufFlux.java:68)
|_ Flux.doOnNext(ByteBufFlux.java:230)
|_ Flux.map(ReactorServerHttpRequest.java:148)
|_ Flux.flatMap(AbstractJackson2Decoder.java:95)
|_ Flux.doFinally(AbstractJackson2Decoder.java:95)
|_ Flux.map(AbstractJackson2Decoder.java:117)
|_ Flux.singleOrEmpty(AbstractJackson2Decoder.java:87)
|_ Operators.error(FluxReceive.java:276)
|_ Mono.onErrorMap(DefaultServerRequest.java:146)
|_ Mono.map(ConferenceService.kt:27)
|_ Mono.map(ConferenceService.kt:32)
|_ Mono.zipWith(ConferenceService.kt:47)
|_ Mono.map(ConferenceService.kt:48)
|_ Flux.flatMap(SimpleReactiveMongoRepository.java:318)
|_ MonoExtensionsKt.toMono(ConferenceService.kt:55)
|_ Mono.map(ConferenceService.kt:56)
|_ Flux.map(AbstractJackson2Encoder.java:99)
|_ Mono.flatMap(DispatcherHandler.java:177)
|_ Mono.onErrorResume(DispatcherHandler.java:177)
|_ Mono.flatMap(DispatcherHandler.java:161)
|_ Mono.defer(DefaultWebFilterChain.java:71)
|_ Mono.doOnSuccess(MetricsWebFilter.java:59)
|_ Mono.doOnError(MetricsWebFilter.java:60)
|_ Mono.compose(MetricsWebFilter.java:54)
|_ Mono.defer(DefaultWebFilterChain.java:71)
|_ Mono.defer(DefaultWebFilterChain.java:71)
我在这里做了一个非常相似的例子：
此路由器将从名称中获取地理位置，并使用另一项服务提取日出和日落时间：
我使用了 Mono 的 and 方法。
internal fun buildResponse(address: Mono<String>) =
address.transform(geoLocationService::fromAddress).and(this::sunriseSunset, ::LocationResponse)
internal fun sunriseSunset(geographicCoordinates: GeographicCoordinates) =
geographicCoordinates.toMono().transform(sunriseSunsetService::fromGeographicCoordinates)
更多详情example
WebClient.exchange() 结果流是单播
这里的问题实际上是 WebClient 每个连接只允许一个订阅者。如果您尝试订阅同一个 exchanged 连接两次 - 您将获得 java.lang.IllegalStateException: Only one connection receive subscriber allowed.
尽管我没有看到您在何处尝试重复使用同一连接两次，但我相信您可以通过使用下一个运算符组合来解决该问题：
class GeoService() {
val client = WebClient.create("https://maps.googleapis.com/maps/api/geocode/")
fun resolveGeoFromCity(city: String): Mono<Geo> {
return client.get()
.uri("json?address=$city&key=$API_KEY&language=en")
.exchange()
.flatMap { it.bodyToMono(String::class.java) }
.map { parse(it) }
.share();
}
...
}
在该示例中，流配置为多播（共享）原始源，只要至少有一个 Subscriber 将被订阅。
如果您需要所有订阅者收到相同的日期，您可以将 .share 替换为 .cache 运算符。
此外，还有一种方法可以替代上述技术。您可以用处理器替换提到的运算符并获得相同的共享可能性：
class GeoService() {
val client = WebClient.create("https://maps.googleapis.com/maps/api/geocode/")
fun resolveGeoFromCity(city: String): Mono<Geo> {
return client.get()
.uri("json?address=$city&key=$API_KEY&language=en")
.exchange()
.flatMap { it.bodyToMono(String::class.java) }
.map { parse(it) }
.subscribeWith(DirectProcessor.create());
}
...
}
在那种情况下，您在调用 subscribeWith 之后就订阅并 运行 使用源的数据，因此，在这种情况下，您可能会丢失部分数据，等等。
为什么 Mono.just(..) 一切正常？
首先.just是一个冷运算符，它允许尽可能多的订阅者在任何时间点接收相同的数据。这就是为什么当您尝试两次从连接中使用相同的数据块时，您没有得到任何异常。
我遇到了类似的问题。解决方法是指定此依赖项：
org.springframework:spring-webflux:5.1.4.RELEASE
当我使用 spring-boot 时，它会部署它的先前版本。不幸的是，link 现在不存在此问题。
所以现在我的 gradle 看起来像这样：
compile('org.springframework.boot:spring-boot-starter-data-mongodb-reactive')
compile('org.springframework.boot:spring-boot-starter-webflux')
// Next 2 dependencies are temporally here until the one above does not resolves next to at least 5.1.4 - where
// webflux issue is resolved:
// https://github.com/rstoyanchev/spr-issue-migration-test-2/issues/17323
compile('org.springframework:spring-webflux:5.1.4.RELEASE')
compile('org.springframework:spring-web:5.1.4.RELEASE')