关于 Spring WebClient 的外部 onTerminate 事件

Question

我是运行 spring-boot v2.0.3 tomcat-embedded webserver 8.5.31，以服务 Spring Webflux REST 服务。其中一个 REST 服务调用另一个外部 REST Web 服务。

public Mono<ServerResponse> select(ServerRequest request) {
  return request.principal().cast(Authentication.class)
      .flatMap(principal ->
          client.get().uri(f -> buildUri(request, principal, request.queryParams(), f))
              .exchange())
      .flatMap((ClientResponse mapper) ->
                  ServerResponse.status(mapper.statusCode())
                      .headers(c -> mapper.headers().asHttpHeaders().forEach(c::put))
                      .body(mapper.bodyToFlux(DataBuffer.class)
                          .delayElements(Duration.ofSeconds(10))
                          .doOnCancel(() -> log.error("Cancelled client"))
                          .doOnTerminate(() -> log.error("Terminated client")), DataBuffer.class))
      .doOnTerminate(() -> log.error("Termination called"));
}

如果浏览器调用我的 REST 服务，并在片刻后取消连接，我可以看到外部 "Termination called" 事件，并且客户端也已终止。但是客户端终止似乎在 tomcat:

中触发错误

2018-07-25 12:50:42.860 DEBUG 12084 --- [      elastic-3] org.example.search.security.UserManager       : Authorizing org.springframework.security.web.authentication.preauth.PreAuthenticatedAuthenticationToken@809aec11: Principal: cn=dv dbsearch client, ou=dbsearch, o=example, l=eb, st=unknown, c=de; Credentials: [PROTECTED]; Authenticated: false; Details: null; Not granted any authorities
2018-07-25 12:50:42.864 DEBUG 12084 --- [      elastic-3] org.example.search.security.UserManager       : Successfully authorized: org.springframework.security.web.authentication.preauth.PreAuthenticatedAuthenticationToken@c03925ec: Principal: org.springframework.security.core.userdetails.User@809aec0e: Username: cn=dv dbsearch client, ou=dbsearch, o=example, l=eb, st=unknown, c=de; Password: [PROTECTED]; Enabled: true; AccountNonExpired: true; credentialsNonExpired: true; AccountNonLocked: true; Granted Authorities: ROLE_ADMIN; Credentials: [PROTECTED]; Authenticated: true; Details: null; Granted Authorities: ROLE_ADMIN
2018-07-25 12:50:45.470 ERROR 12084 --- [ctor-http-nio-4] c.d.s.s.h.SolrSelectRequestHandler       : Termination called
2018-07-25 12:51:15.562 ERROR 12084 --- [     parallel-3] c.d.s.s.h.SolrSelectRequestHandler       : Terminated client
2018-07-25 12:51:15.625 ERROR 12084 --- [nio-8443-exec-2] o.s.w.s.adapter.HttpWebHandlerAdapter    : Unhandled failure: Eine bestehende Verbindung wurde softwaregesteuert durch den Hostcomputer abgebrochen, response already set (status=200)
2018-07-25 12:51:15.628  WARN 12084 --- [nio-8443-exec-2] o.s.h.s.r.ServletHttpHandlerAdapter      : Handling completed with error: Eine bestehende Verbindung wurde softwaregesteuert durch den Hostcomputer abgebrochen
2018-07-25 12:51:15.652 ERROR 12084 --- [nio-8443-exec-2] o.a.catalina.connector.CoyoteAdapter     : Exception while processing an asynchronous request

  java.lang.IllegalStateException: Calling [asyncError()] is not valid for a request with Async state [DISPATCHING]
  at org.apache.coyote.AsyncStateMachine.asyncError(AsyncStateMachine.java:424)
  at org.apache.coyote.AbstractProcessor.action(AbstractProcessor.java:470)
  at org.apache.coyote.Request.action(Request.java:431)
  at org.apache.catalina.core.AsyncContextImpl.setErrorState(AsyncContextImpl.java:388)
  at org.apache.catalina.connector.CoyoteAdapter.asyncDispatch(CoyoteAdapter.java:176)
  at org.apache.coyote.AbstractProcessor.dispatch(AbstractProcessor.java:232)
  at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:53)
  at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:790)
  at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1468)
  at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
  at java.lang.Thread.run(Thread.java:748)

抱歉出现德语错误消息，这意味着 "client abortet connection"。

我对这个错误消息本身并没有真正的问题，只是，我在 spring 的 Webclient 中的缓冲区似乎没有被清除（我没有在本地重现的日志, 所以它有不同的时间戳):

2018-07-23 08:44:36.892 ERROR 22707 — [reactor-http-nio-5] io.netty.util.ResourceLeakDetector       : LEAK: ByteBuf.release() was not called before it's garbage-collected. See http://netty.io/wiki/reference-counted-objects.html for more information.

Recent access records:

Created at:
   io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:331)      io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:185)

所以这里的问题是：当对我的 REST 服务的请求被取消时，我如何干净地结束 WebClient 连接？

Answer 1

我不能确定那个异常消息，但我知道 Tomcat 在第 8.5.x 代中对此进行了改进。您使用的是哪个版本？如果您可以提供一种一致的方法来使用最小的应用程序重现此问题，则可以在 Spring Framework 上的 jira.spring.io 中创建一个新问题，或者 Tomcat 本身，如果您设法在没有 Spring（虽然它应该很难重现）。

现在关于释放 DataBuffer 个实例 - DataBuffer 个实例可以合并，具体取决于实现。这里 WebClient 使用的是 Netty，它是池化缓冲区。所以当它们不再被使用时需要被释放。

看看你的实现，我认为那些未释放的缓冲区来自于：

WebClient 正在从远程端点获取数据并创建 DataBuffer 个实例
沿途的各种 Reactor 运算符正在缓冲那些使用内部队列的运算符（根据预取和使用的运算符，排队缓冲区的数量可能会有所不同）
当订阅者失败或取消时，那些位于内部队列中的缓冲区不会按应有的方式释放。

目前 Reactor 不提供挂钩点来访问那些错误情况下的那些对象。但这是 Reactor core 3.2.0 中添加的全新功能。 Spring Framework 和 SPR-17025 将在内部利用这一点。请关注此问题 - 在测试修复时，您的用例可能会很方便。

关于 Spring WebClient 的外部 onTerminate 事件

About Spring WebClient on external onTerminate event

spring-boot

project-reactor

spring-webflux