当Tomcat遇上Netty(续集)

本篇接着上篇，主要讲一下Tomcat与Netty是怎么勾搭上的，过程有点复杂。

Tomcat与Netty是如何衔接起来的？

请看下面这张图：

从下往下看，接收请求的时候走的确实是tomcat，然后通过spring cloud gateway的过滤器链，走到了一个叫作 NettyWriteResponseFilter的过滤器。

再接着往下走，又走到了一个叫作 NettyRoutingFilter的一个过滤器，这个过滤器是干什么的呢？

从名字可以看出，它是用来做路由的，也就是在这里把gateway接收到的请求转发给其它服务。

这里路由的实现其实就是创建一个HttpClient，然后根据配置的路由信息拿到目标地址，然后再向目标服务发送一个Http请求。

我们简单地看一下NettyRoutingFilter里面的实现（跟着我的注释看就好）：

1. publicMono<Void> filter(ServerWebExchange exchange,GatewayFilterChain chain){

2. // 拿到目标地址，在前面放进去的

3. URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);

4. 
   

5. String scheme = requestUrl.getScheme();

6. if(isAlreadyRouted(exchange)

7. ||(!"http".equals(scheme)&&!"https".equals(scheme))){

8. return chain.filter(exchange);

9. }

10. setAlreadyRouted(exchange);

11. 
    

12. // 原始请求

13. ServerHttpRequest request = exchange.getRequest();

14. 
    

15. finalHttpMethod method =HttpMethod.valueOf(request.getMethodValue());

16. finalString url = requestUrl.toASCIIString();

17. 
    

18. HttpHeaders filtered = filterRequest(getHeadersFilters(), exchange);

19. 
    

20. // 构造新的请求头

21. finalDefaultHttpHeaders httpHeaders =newDefaultHttpHeaders();

22. filtered.forEach(httpHeaders::set);

23. 
    

24. boolean preserveHost = exchange

25. .getAttributeOrDefault(PRESERVE_HOST_HEADER_ATTRIBUTE,false);

26. Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR);

27. 
    

28. // 获取一个HttpClient

29. // key1, 此时还在tomcat的线程里

30. Flux<HttpClientResponse> responseFlux = getHttpClient(route, exchange)

31. // 设置请求头

32. .headers(headers ->{

33. headers.add(httpHeaders);

34. // Will either be set below, or later by Netty

35. headers.remove(HttpHeaders.HOST);

36. if(preserveHost){

37. String host = request.getHeaders().getFirst(HttpHeaders.HOST);

38. headers.add(HttpHeaders.HOST, host);

39. }

40. // 发送请求，本文来源于工纵耗彤哥读源码

41. }).request(method).uri(url).send((req, nettyOutbound)->{

42. if(log.isTraceEnabled()){

43. nettyOutbound

44. .withConnection(connection -> log.trace("outbound route: "

45. + connection.channel().id().asShortText()

46. +", inbound: "+ exchange.getLogPrefix()));

47. }

48. // 发送请求

49. // key2，此时已经到netty的线程里了

50. return nettyOutbound.send(request.getBody().map(this::getByteBuf));

51. }).responseConnection((res, connection)->{

52. // response的处理

53. 
    

54. // Defer committing the response until all route filters have run

55. // Put client response as ServerWebExchange attribute and write

56. // response later NettyWriteResponseFilter

57. exchange.getAttributes().put(CLIENT_RESPONSE_ATTR, res);

58. exchange.getAttributes().put(CLIENT_RESPONSE_CONN_ATTR, connection);

59. 
    

60. ServerHttpResponse response = exchange.getResponse();

61. // put headers and status so filters can modify the response

62. HttpHeaders headers =newHttpHeaders();

63. 
    

64. res.responseHeaders().forEach(

65. entry -> headers.add(entry.getKey(), entry.getValue()));

66. 
    

67. String contentTypeValue = headers.getFirst(HttpHeaders.CONTENT_TYPE);

68. if(StringUtils.hasLength(contentTypeValue)){

69. exchange.getAttributes().put(ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR,

70. contentTypeValue);

71. }

72. 
    

73. setResponseStatus(res, response);

74. 
    

75. // make sure headers filters run after setting status so it is

76. // available in response

77. HttpHeaders filteredResponseHeaders =HttpHeadersFilter.filter(

78. getHeadersFilters(), headers, exchange,Type.RESPONSE);

79. 
    

80. if(!filteredResponseHeaders

81. .containsKey(HttpHeaders.TRANSFER_ENCODING)

82. && filteredResponseHeaders

83. .containsKey(HttpHeaders.CONTENT_LENGTH)){

84. // It is not valid to have both the transfer-encoding header and

85. // the content-length header.

86. // Remove the transfer-encoding header in the response if the

87. // content-length header is present.

88. response.getHeaders().remove(HttpHeaders.TRANSFER_ENCODING);

89. }

90. 
    

91. exchange.getAttributes().put(CLIENT_RESPONSE_HEADER_NAMES,

92. filteredResponseHeaders.keySet());

93. 
    

94. response.getHeaders().putAll(filteredResponseHeaders);

95. 
    

96. returnMono.just(res);

97. });

98. 
    

99. Duration responseTimeout = getResponseTimeout(route);

100. if(responseTimeout !=null){

101. responseFlux = responseFlux

102. .timeout(responseTimeout,Mono.error(newTimeoutException(

103. "Response took longer than timeout: "+ responseTimeout)))

104. .onErrorMap(TimeoutException.class,

105. th ->newResponseStatusException(HttpStatus.GATEWAY_TIMEOUT,

106. th.getMessage(), th));

107. }

108. 
     

109. // 过滤器链

110. return responseFlux.then(chain.filter(exchange));

111. }

忽略代码本身的复杂性，整体逻辑就是向目标服务发送请求，并处理响应体。

但是，这里真正发请求的时候是在Netty线程里发送的，同时，处理响应体也同样是在Netty线程中进行的。

而在处理响应之前呢，当然是收到目标服务的响应，在接收目标服务的响应的时候就是在Netty的NioEventLoop中进行的，Netty接收到响应后会创建一个ByteBuf来承载响应的内容，最后，经过一系列的调用就回到了NettyWriteResponseFilter的回调里，在NettyWriteResponseFilter里对响应体进行写出操作，让我们看一下这个类里面的基本内容：

1. publicMono<Void> filter(ServerWebExchange exchange,GatewayFilterChain chain){

2. // NOTICE: nothing in "pre" filter stage as CLIENT_RESPONSE_CONN_ATTR is not added

3. // until the NettyRoutingFilter is run

4. // @formatter:off

5. return chain.filter(exchange)

6. .doOnError(throwable -> cleanup(exchange))

7. .then(Mono.defer(()->{

8. // 响应的回调

9. 
   

10. Connection connection = exchange.getAttribute(CLIENT_RESPONSE_CONN_ATTR);

11. 
    

12. if(connection ==null){

13. returnMono.empty();

14. }

15. if(log.isTraceEnabled()){

16. log.trace("NettyWriteResponseFilter start inbound: "

17. + connection.channel().id().asShortText()+", outbound: "

18. + exchange.getLogPrefix());

19. }

20. // 响应

21. ServerHttpResponse response = exchange.getResponse();

22. 
    

23. // TODO: needed?

24. finalFlux<DataBuffer> body = connection

25. .inbound()

26. .receive()

27. .retain()

28. .map(byteBuf -> wrap(byteBuf, response));

29. 
    

30. MediaType contentType =null;

31. try{

32. contentType = response.getHeaders().getContentType();

33. }

34. catch(Exception e){

35. if(log.isTraceEnabled()){

36. log.trace("invalid media type", e);

37. }

38. }

39. 
    

40. // 写出，本文来源于工纵耗彤哥读源码

41. return(isStreamingMediaType(contentType)

42. ? response.writeAndFlushWith(body.map(Flux::just))

43. : response.writeWith(body));

44. })).doOnCancel(()-> cleanup(exchange));

45. // @formatter:on

46. }

但是，这里的response是TomcatServerHttpResponse，因为接收请求是通过tomcat接收的，所以，这里的响应是tomcat的。

因此，最后是调用了tomcat的write()方法或者writeAndFlush()方法，此时，已经没有Netty什么事了。

在整个过程中，接收目标服务的响应的时候通过Netty分配了ByteBuf而把这个响应返回给调用者的时候却是走的tomcat，导致这个分配的ByteBuf一直没有释放，所以，出现了内存泄漏。

如果全程都使用Netty的情况下，也会经历上面说到的这些步骤，只不过在最后这里的响应会变成ReactorServerHttpResponse，而不是tomcat的响应。

在ReactorServerHttpResponse里面就会调用到Netty的相关方法，并往Netty的线程池里放一个任务，这个任务是reactor.netty.channel.MonoSendMany.SendManyInner.AsyncFlush：

1. finalclassAsyncFlushimplementsRunnable{

2. @Override

3. publicvoid run(){

4. if(pending !=0){

5. ctx.flush();

6. }

7. }

8. }

在这个任务里面调用ctx.flush()，这个flush()就是把内容真正地发送出去的方法，发送完了之后，也会把相应的ByteBuf给清理了，也就释放了内存。

好了，让我们用一张图来表示一下整个的过程：

另外，我给spring cloud gateway官方提了一个issue：https://github.com/spring-cloud/spring-cloud-gateway/issues/1728

但是，这官方有点扯淡，直接把我的issue关了，说不支持tomcat，但是确实能运行起来，而且跑得很好，直到出现内存溢出为止，呵呵了。

最后，我把调试的源码放在github上了，你也可以下载下来自己试一下，地址如下：https://github.com/alan-tang-tt/springcloud-gateway-oom。