防止内存持续性增长, 异常场景下server端netty堆外内存释放

cat-core/src/main/java/com/dianping/cat/analysis/DefaultMessageHandler.java

@@ -1,29 +1,30 @@
-/*
- * Copyright (c) 2011-2018, Meituan Dianping. All Rights Reserved.
- *
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
+/*
+ * Copyright (c) 2011-2018, Meituan Dianping. All Rights Reserved.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package com.dianping.cat.analysis;
 
-import org.codehaus.plexus.logging.LogEnabled;
-import org.codehaus.plexus.logging.Logger;
-import org.unidal.lookup.ContainerHolder;
-import org.unidal.lookup.annotation.Inject;
-import org.unidal.lookup.annotation.Named;
-
+import com.dianping.cat.Cat;
+import org.codehaus.plexus.logging.LogEnabled;
+import org.codehaus.plexus.logging.Logger;
+import org.unidal.lookup.ContainerHolder;
+import org.unidal.lookup.annotation.Inject;
+import org.unidal.lookup.annotation.Named;
+
 import com.dianping.cat.message.spi.MessageTree;
 
 @Named(type = MessageHandler.class)
@@ -48,6 +49,12 @@ public void handle(MessageTree tree) {
 			m_consumer.consume(tree);
 		} catch (Throwable e) {
 			m_logger.error("Error when consuming message in " + m_consumer + "! tree: " + tree, e);
+			//2024-08-12: 消息丢弃时 释放堆外内存 防止内存泄露
+			try {
+				tree.close();
+			}catch (Exception ex){
+				//Cat.logError(ex);
+			}
 		}
 	}
 }

cat-core/src/main/java/com/dianping/cat/analysis/RealtimeConsumer.java

@@ -60,6 +60,12 @@ public void consume(MessageTree tree) {
 			period.distribute(tree);
 		} else {
 			m_serverStateManager.addNetworkTimeError(1);
+			//2024-08-12: 消息丢弃时 释放堆外内存 防止内存泄露
+			try {
+				tree.close();
+			}catch (Exception ex){
+				//Cat.logError(ex);
+			}
 		}
 	}
 

cat-core/src/main/java/com/dianping/cat/analysis/TcpSocketReceiver.java

@@ -181,6 +181,11 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> ou
 			} catch (Exception e) {
 				m_serverStateManager.addMessageTotalLoss(1);
 				m_logger.error(e.getMessage(), e);
+
+				//2024-08-12: 消息丢弃时 释放堆外内存 防止内存泄露
+				if(buffer != null){
+					BufReleaseHelper.release(buffer);
+				}
 			}
 		}
 	}

cat-core/src/main/java/com/dianping/cat/message/spi/DefaultMessageTree.java

@@ -18,6 +18,7 @@
  */
 package com.dianping.cat.message.spi;
 
+import java.io.IOException;
 import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.List;
@@ -361,5 +362,36 @@ public String toString() {
         return result;
     }
 
+	//2024-08-13: 消息丢弃时 释放堆外内存 防止内存泄露
+	private volatile boolean isClosed = false;
+
+	@Override
+	public void close() throws IOException {
+		try {
+			// 显式关闭资源
+			// 其他关闭操作
+			if (!isClosed && this.getBuffer() != null) {
+				BufReleaseHelper.release(this.getBuffer());
+			}
+			isClosed = true;
+		} catch (Exception ex) {
+//			Cat.logError(ex);
+		} finally {
+
+		}
+	}
+
+	@Override
+	protected void finalize() throws Throwable {
+		super.finalize();
+		if (!isClosed) {
+			try {
+				this.close(); // 尝试关闭资源
+			} catch (Exception e) {
+				// 处理关闭时可能抛出的异常
+				Cat.logError(e);
+			}
+		}
+	}
 
 }

cat-core/src/main/java/com/dianping/cat/message/spi/MessageTree.java

@@ -1,35 +1,36 @@
-/*
- * Copyright (c) 2011-2018, Meituan Dianping. All Rights Reserved.
- *
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
+/*
+ * Copyright (c) 2011-2018, Meituan Dianping. All Rights Reserved.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package com.dianping.cat.message.spi;
 
-import java.util.List;
-
-import com.dianping.cat.message.Event;
-import com.dianping.cat.message.Heartbeat;
-import com.dianping.cat.message.Message;
-import com.dianping.cat.message.Metric;
-import com.dianping.cat.message.Transaction;
-import com.dianping.cat.message.tree.MessageId;
-
+import java.io.Closeable;
+import java.util.List;
+
+import com.dianping.cat.message.Event;
+import com.dianping.cat.message.Heartbeat;
+import com.dianping.cat.message.Message;
+import com.dianping.cat.message.Metric;
+import com.dianping.cat.message.Transaction;
+import com.dianping.cat.message.tree.MessageId;
+
 import io.netty.buffer.ByteBuf;
 
-public interface MessageTree extends Cloneable {
+public interface MessageTree extends Cloneable, Closeable {
 	public boolean canDiscard();
 
 	public MessageTree copy();

cat-hadoop/src/main/java/org/unidal/cat/message/storage/internals/DefaultMessageDumper.java

@@ -160,6 +160,12 @@ public void process(MessageTree tree) {
 				m_logger.info("message tree queue is full " + m_failCount + " index " + index);
 				// tree.getBuffer().release();
 			}
+			//2024-08-12: 消息丢弃时 释放堆外内存 防止内存泄露
+			try {
+				tree.close();
+			}catch (Exception ex){
+				//Cat.logError(ex);
+			}
 		} else {
 			m_statisticManager.addMessageSize(domain, tree.getBuffer().readableBytes());
 

cat-hadoop/src/main/java/org/unidal/cat/message/storage/internals/DefaultMessageProcessor.java

@@ -136,7 +136,13 @@ private void processMessage(MessageTree tree) {
 		} catch (Exception e) {
 			Cat.logError(e);
 		} finally {
-			ReferenceCountUtil.release(buffer);
+//			ReferenceCountUtil.release(buffer);
+			//2024-08-12: 消息丢弃时 释放堆外内存 防止内存泄露
+			try {
+				tree.close();
+			}catch (Exception ex){
+				//Cat.logError(ex);
+			}
 		}
 	}