java - 泄漏:未调用 ByteBuf.release() - 我们如何解决这个问题?
问题描述
我们有一个基于网络的网络流量密集型 Java 应用程序/服务器。
旁注:我主要支持这个应用程序,我没有构建它,所以我不完全了解它。
我们有时会收到此错误,如下所示。以前我们通常在服务器启动 3-4 天后收到此错误。现在我注意到我们在重新启动服务器/应用程序后仅 10-15 分钟就收到了这个错误。
我不明白这怎么可能。这个错误是否令人担忧,我们该如何解决?我记得过去对同样的错误进行了广泛的研究,当时我什至尝试升级和修补 netty,但没有任何帮助完全解决问题。
操作系统:Linux
Java 版本:1.8
Netty 版本:netty-all-4.1.30.Final.jar
这是唯一一行特定于应用程序的代码,其他一切都发生在 Netty 中。
com.company.japp.protocol.http.decoders.ConditionalHttpChunkAggregator.channelRead
这是 Netty 本身的某种错误吗?Netty 升级或任何其他配置调整在这里有帮助吗?
[2020-09-04 08:33:53,072]
ERROR
io.netty.util.ResourceLeakDetector
LEAK: ByteBuf.release() was not called before it's garbage-collected.
See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records:
Created at:
io.netty.buffer.AbstractByteBufAllocator.compositeDirectBuffer(AbstractByteBufAllocator.java:221)
io.netty.buffer.AbstractByteBufAllocator.compositeBuffer(AbstractByteBufAllocator.java:199)
io.netty.handler.codec.MessageAggregator.decode(MessageAggregator.java:255)
io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:88)
com.company.japp.protocol.http.decoders.ConditionalHttpChunkAggregator.channelRead(ConditionalHttpChunkAggregator.java:112)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:323)
io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:297)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1434)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:965)
io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163)
io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:644)
io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:579)
io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:496)
io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:458)
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:897)
io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
java.lang.Thread.run(Thread.java:748)
这是ConditionalHttpChunkAggregator
.
package com.company.japp.protocol.http.decoders;
import com.company.japp.IHttpProxyServer;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpMessage;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import java.util.HashSet;
@ChannelHandler.Sharable
public class ConditionalHttpChunkAggregator extends HttpObjectAggregator {
private static final InternalLogger logger = InternalLoggerFactory.getInstance(ConditionalHttpChunkAggregator.class);
private volatile boolean sendaschunked;
private volatile int maxContentLength;
private static IHttpProxyServer iHttpProxyServer;
public static void initialize(IHttpProxyServer iHttpProxyServer) {
ConditionalHttpChunkAggregator.iHttpProxyServer = iHttpProxyServer;
}
public ConditionalHttpChunkAggregator(int maxContentLength) {
super(maxContentLength);
this.maxContentLength = maxContentLength;
sendaschunked = false;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg){
if ((msg instanceof HttpResponse)) {
HttpResponse response = (HttpResponse)msg;
if ((msg instanceof HttpMessage)) {
HttpMessage httpmessage= (HttpMessage)msg;
try {
// If the content length exceeds the threshhold, then send it as chunked
// It's too large to process substitutions
Long contentlength =
httpmessage.headers().get(HttpHeaders.Names.CONTENT_LENGTH) != null ?
Long.valueOf(httpmessage.headers().get(HttpHeaders.Names.CONTENT_LENGTH)) : -1;
if (contentlength >= maxContentLength) {
sendaschunked = true;
} else {
// Check content types
HashSet<String> chunkabletypes = iHttpProxyServer.getConfig().getProperty("chunkabletypes");
if (!chunkabletypes.isEmpty() && response.headers().contains(HttpHeaders.Names.CONTENT_TYPE)) {
String contentType = response.headers().get(HttpHeaders.Names.CONTENT_TYPE).toLowerCase().trim();
if (contentType.length()>0) {
sendaschunked = chunkabletypes.contains(contentType);
if (!sendaschunked) {
for (String chunkabletype: chunkabletypes) {
// Begins with
if (contentType.indexOf(chunkabletype)==0) {
sendaschunked = true;
break;
}
}
}
}
}
}
if (sendaschunked) {
ctx.fireChannelRead(msg);
return;
}
}
catch(Exception ex) {
logger.error("error determining chunkable viability", ex);
}
}
}
if (sendaschunked) {
ctx.fireChannelRead(msg);
return;
}
try {
super.channelRead(ctx, msg);
} catch (Exception e) {
logger.error("error determining chunkable viability", e);
e.printStackTrace();
}
}
}
这是可分块类型属性的值:
video/x-ms-wvx,video/x-flv,application/x-shockwave-flash,video/quicktime,video/,audio/
编辑
我认为这是 netty 4.1.30 中 255 行中的错误io.netty.handler.codec.MessageAggregator
。似乎这个 CompositeByteBuf 已分配但未释放。我对吗?我希望有一些权威的答案来确认或拒绝这个想法。
// A streamed message - initialize the cumulative buffer, and wait for incoming chunks.
CompositeByteBuf content = ctx.alloc().compositeBuffer(maxCumulationBufferComponents); // LINE 255
if (m instanceof ByteBufHolder) {
appendPartialContent(content, ((ByteBufHolder) m).content());
}
currentMessage = beginAggregation(m, content);
解决方案
在我们的例子中,响应正文中的一个问题是没有被读取,导致字节 buf 永远不会被释放。
原始代码:
WebClient client = WebClient.create("some_url");
return client.get().exchange().flatMap(response -> Mono.just(builder.up().build()));
固定代码:
WebClient client = WebClient.create("some_url");
return client.get().retrieve().toBodilessEntity().flatMap(response -> Mono.just(builder.up().build()));
甚至 javadoc forexchange
也这么说:
由于可能泄漏内存和/或连接,自 5.3 起已弃用;请使用 exchangeToMono(Function), exchangeToFlux(Function); 还可以考虑使用 retrieve(),它通过 ResponseEntity 提供对响应状态和标头的访问以及错误状态处理。
荣誉:https ://stackoverflow.com/a/51321602/3242721
为了在 localhost 上重现该问题,我使用了-Dio.netty.leakDetectionLevel=paranoid
vm args。
推荐阅读
- angular - Angular Typescript声纳错误必须使用“过滤器”的返回值
- c# - dotnet-dump 仅限于 .net 核心应用程序?
- javascript - mb_strlen() 计算新行两次 - 与 JavaScript 中的 .length() 不同
- css - 导入作用域引导 css
- firebase - 如何从firebase获取特定的id数据作为地图?
- android - Android Studio 和约束布局编辑器问题
- javascript - Javascript 日期循环
- python - 在 Python 中,所有基本数据类型都是不可变的,为什么在 String Replace() 函数中有效?
- apache-kafka-streams - KStreams 关闭 Aggregate 的内部主题
- java - Java - 客户端-服务器程序 - http 响应