Python中文网

一个关于 编程问题的解答网站.

有 Java 编程相关的问题?

你可以在下面搜索框中键入要查询的问题!

java WebClient请求级别超时抛出名为default onErrorDropped的运算符

1 月 Questions & Answers

我们正在使用Spring反应式WebClient进行http调用。它在下面使用JettyClientHttpConnector。为了设置请求级别的超时，我们使用Mono timeout API

如果服务器没有响应，mono会超时，但也会抛出

Operator called default onErrorDropped java.lang.InterruptedException

在这个错误的几个实例之后，我们在jetty连接池中遇到了线程不足的问题

java.lang.IllegalStateException: Insufficient configured threads: required=200 \u003c max=200 for QueuedThreadPool[HttpClient@49e4c2d5]@687d1782{STARTED,8\u003c=199\u003c=200,i=1,q=0}[ReservedThreadExecutor@3eaa0b62{s=0/1,p=0}]

我们怀疑由于异常，连接没有被释放。如何确保在Mono超时后释放连接资源

依赖关系

org.springframework:spring-webflux:jar:5.1.4.RELEASE
org.springframework.boot:spring-boot-starter-jetty:jar:2.1.2.RELEASE
org.springframework.cloud:spring-cloud-stream-reactive:jar:2.2.0.M1

网络客户端配置

WebClient.Builder.clone()
.clientConnector(new JettyClientHttpConnector(new HttpClient())).build();

客户电话


URI request = UriComponentsBuilder
.fromUriString("foo")
.path("/path")
.build();

return webClient.put()
   .uri(request)
   .body(BodyInserters.fromObject(fooBar))
   .retrieve()
   .onStatus(HttpStatus::isError, clientResponse -> 
 clientResponse.toEntity(String.class)
                .map(body -> new RunTimeException(body.toString())))
   .bodyToMono(Foo2.class)
   .timeout(Duration.ofMillis(100))
   .retry(3, retryError -> retryError instanceof TimeoutException);

例外情况

单声道超时：-

java.util.concurrent.TimeoutException: Did not observe any item or terminal signal within 500ms in 'lift' (and no fallback has been configured)

reactor.core.publisher.Operators         : Operator called default onErrorDropped java.lang.InterruptedException: null
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998) ~[na:1.8.0_192]
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) ~[na:1.8.0_192]
        at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231) ~[na:1.8.0_192]
        at org.eclipse.jetty.io.ManagedSelector.doStart(ManagedSelector.java:106) ~[jetty-io-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.ContainerLifeCycle.start(ContainerLifeCycle.java:138) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.ContainerLifeCycle.doStart(ContainerLifeCycle.java:108) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.io.SelectorManager.doStart(SelectorManager.java:262) ~[jetty-io-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.ContainerLifeCycle.start(ContainerLifeCycle.java:138) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.ContainerLifeCycle.doStart(ContainerLifeCycle.java:108) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.client.AbstractConnectorHttpClientTransport.doStart(AbstractConnectorHttpClientTransport.java:64) ~[jetty-client-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.ContainerLifeCycle.start(ContainerLifeCycle.java:138) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.ContainerLifeCycle.doStart(ContainerLifeCycle.java:108) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.client.HttpClient.doStart(HttpClient.java:250) ~[jetty-client-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.springframework.http.client.reactive.JettyClientHttpConnector.connect(JettyClientHttpConnector.java:98) ~[spring-web-5.1.4.RELEASE.jar!/:5.1.4.RELEASE]
        at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.exchange(ExchangeFunctions.java:103) ~[spring-webflux-5.1.4.RELEASE.jar!/:5.1.4.RELEASE]
        at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:118) ~[reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:96) [spring-cloud-sleuth-core-2.1.1.RELEASE.jar!/:2.1.1.RELEASE]
        at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:121) [reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:96) [spring-cloud-sleuth-core-2.1.1.RELEASE.jar!/:2.1.1.RELEASE]
        at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505) [reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241) [reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:96) [spring-cloud-sleuth-core-2.1.1.RELEASE.jar!/:2.1.1.RELEASE]
        at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onNext(MonoSubscribeOn.java:143) [reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:96) [spring-cloud-sleuth-core-2.1.1.RELEASE.jar!/:2.1.1.RELEASE]
        at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505) [reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.publisher.MonoCallable.subscribe(MonoCallable.java:61) [reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.publisher.MonoLiftFuseable.subscribe(MonoLiftFuseable.java:55) [reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.publisher.Mono.subscribe(Mono.java:3695) [reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:123) [reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) [reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) [reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at org.springframework.cloud.sleuth.instrument.async.TraceCallable.call(TraceCallable.java:70) [spring-cloud-sleuth-core-2.1.1.RELEASE.jar!/:2.1.1.RELEASE]
        at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_192]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [na:1.8.0_192]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [na:1.8.0_192]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_192]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_192]
        at java.lang.Thread.run(Thread.java:748) [na:1.8.0_192]

reactor.core.scheduler.Schedulers        : Scheduler worker in group main failed with an uncaught exception java.lang.InterruptedException: null
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998) ~[na:1.8.0_192]
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) ~[na:1.8.0_192]
        at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231) ~[na:1.8.0_192]
        at org.eclipse.jetty.io.ManagedSelector.doStart(ManagedSelector.java:106) ~[jetty-io-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.ContainerLifeCycle.start(ContainerLifeCycle.java:138) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.ContainerLifeCycle.doStart(ContainerLifeCycle.java:108) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.io.SelectorManager.doStart(SelectorManager.java:262) ~[jetty-io-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.ContainerLifeCycle.start(ContainerLifeCycle.java:138) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.ContainerLifeCycle.doStart(ContainerLifeCycle.java:108) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.client.AbstractConnectorHttpClientTransport.doStart(AbstractConnectorHttpClientTransport.java:64) ~[jetty-client-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.ContainerLifeCycle.start(ContainerLifeCycle.java:138) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.ContainerLifeCycle.doStart(ContainerLifeCycle.java:108) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.client.HttpClient.doStart(HttpClient.java:250) ~[jetty-client-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68) ~[jetty-util-9.4.14.v20181114.jar!/:9.4.14.v20181114]
        at org.springframework.http.client.reactive.JettyClientHttpConnector.connect(JettyClientHttpConnector.java:98) ~[spring-web-5.1.4.RELEASE.jar!/:5.1.4.RELEASE]
        at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.exchange(ExchangeFunctions.java:103) ~[spring-webflux-5.1.4.RELEASE.jar!/:5.1.4.RELEASE]
        at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:118) ~[reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:96) ~[spring-cloud-sleuth-core-2.1.1.RELEASE.jar!/:2.1.1.RELEASE]
        at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:121) ~[reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:96) ~[spring-cloud-sleuth-core-2.1.1.RELEASE.jar!/:2.1.1.RELEASE]
        at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505) ~[reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241) ~[reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:96) ~[spring-cloud-sleuth-core-2.1.1.RELEASE.jar!/:2.1.1.RELEASE]
        at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onNext(MonoSubscribeOn.java:143) ~[reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onNext(ScopePassingSpanSubscriber.java:96) ~[spring-cloud-sleuth-core-2.1.1.RELEASE.jar!/:2.1.1.RELEASE]
        at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505) ~[reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.publisher.MonoCallable.subscribe(MonoCallable.java:61) ~[reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.publisher.MonoLiftFuseable.subscribe(MonoLiftFuseable.java:55) ~[reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.publisher.Mono.subscribe(Mono.java:3695) ~[reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:123) ~[reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) [reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) [reactor-core-3.2.5.RELEASE.jar!/:3.2.5.RELEASE]
        at org.springframework.cloud.sleuth.instrument.async.TraceCallable.call(TraceCallable.java:70) ~[spring-cloud-sleuth-core-2.1.1.RELEASE.jar!/:2.1.1.RELEASE]
        at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0_192]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) ~[na:1.8.0_192]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) ~[na:1.8.0_192]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[na:1.8.0_192]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[na:1.8.0_192]
        at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_192]

共 (1) 个答案

  1. # 1 楼答案

    这类错误（onXxxDropped）代表“协议错误”，其中接收到的信号不符合反应流规范。在这里，一个onError信号（该InterruptedException）在另一个onError（超时）之后接收

    我不确定是什么导致了中断（jetty或webflux），也不确定是否可以在WebClient中修复根本原因，但您可以注册一个全局钩子来更改默认的onErrorDropped行为（请参见Hooks#onErrorDropped(Consumer)）

    或者，您可以使用以下（不受支持的）代码在有问题的Flux上注册一个本地钩子：

    Flux<T> doubleErroringFlux;
Consumer<Throwable> droppedErrorConsumer = ...;
Flux<T> fluxToUser = doubleErroringFlux
    .subscriberContext(Context.of("reactor.onErrorDropped.local",
    droppedErrorConsumer);
fluxToUse.subscribe(); //or pass down the method, or whatever