# 經(jīng)典內(nèi)存泄漏示例
class DataCache:
    def __init__(self):
        self._cache = {}  # 永遠不清理的緩存
    
    def add_data(self, key, value):
        self._cache[key] = value  # 數(shù)據(jù)只增不減
    
    def process_request(self, request_id, data):
        # 每個請求都緩存數(shù)據(jù)，從不刪除
        self.add_data(request_id, data)
        return f"Processed {request_id}"

# 使用示例
cache = DataCache()
for i in range(1000000):
    # 一百萬次請求后，內(nèi)存爆炸！
    cache.process_request(f"req_{i}", "x" * 1000)

import sys

# 引用計數(shù)示例
obj = [1, 2, 3]
print(f"初始引用計數(shù): {sys.getrefcount(obj) - 1}")  # -1 因為 getrefcount 自己也引用了

ref1 = obj
print(f"增加引用后: {sys.getrefcount(obj) - 1}")

del ref1
print(f"刪除引用后: {sys.getrefcount(obj) - 1}")

# 循環(huán)引用問題
class Node:
    def __init__(self, value):
        self.value = value
        self.next = None

# 創(chuàng)建循環(huán)引用
node1 = Node(1)
node2 = Node(2)
node1.next = node2
node2.next = node1  # 循環(huán)！

# 即使刪除引用，循環(huán)內(nèi)的對象也不會立即釋放
del node1, node2
# GC 會在后臺處理，但可能有延遲

# 場景一：全局容器無限增長
global_logs = []

def log_event(event):
    global_logs.append(event)  # 永不清理

# 場景二：閉包捕獲大對象
def create_handler(large_data):
    def handler():
        # 閉包持有 large_data 引用
        return len(large_data)
    return handler

# 場景三：未正確關(guān)閉資源
class FileProcessor:
    def __init__(self, filename):
        self.file = open(filename)  # 沒有 __del__ 或 __exit__
    
    def process(self):
        return self.file.read()

# 場景四：緩存未設(shè)置過期策略
cache = {}
def get_or_compute(key):
    if key not in cache:
        cache[key] = expensive_computation(key)
    return cache[key]

def expensive_computation(key):
    return [0] * 1000000  # 模擬大對象

import tracemalloc
import linecache

def display_top_memory(snapshot, key_type='lineno', limit=10):
    """顯示內(nèi)存占用 Top N"""
    snapshot = snapshot.filter_traces((
        tracemalloc.Filter(False, "<frozen importlib._bootstrap>"),
        tracemalloc.Filter(False, "<unknown>"),
    ))
    
    top_stats = snapshot.statistics(key_type)
    
    print(f"\n{'='*70}")
    print(f"Top {limit} 內(nèi)存占用（按 {key_type} 排序）")
    print(f"{'='*70}")
    
    for index, stat in enumerate(top_stats[:limit], 1):
        frame = stat.traceback[0]
        filename = frame.filename
        lineno = frame.lineno
        
        # 獲取源代碼
        line = linecache.getline(filename, lineno).strip()
        
        print(f"\n#{index}: {filename}:{lineno}")
        print(f"    {line}")
        print(f"    大小: {stat.size / 1024 / 1024:.1f} MB")
        print(f"    數(shù)量: {stat.count} 個對象")

# 實戰(zhàn)案例：檢測內(nèi)存泄漏
def memory_leak_example():
    """模擬內(nèi)存泄漏"""
    tracemalloc.start()
    
    # 快照 1：初始狀態(tài)
    snapshot1 = tracemalloc.take_snapshot()
    
    # 執(zhí)行可能泄漏的代碼
    leaked_objects = []
    for i in range(10000):
        # 故意泄漏：創(chuàng)建對象但不釋放
        leaked_objects.append([0] * 1000)
    
    # 快照 2：執(zhí)行后狀態(tài)
    snapshot2 = tracemalloc.take_snapshot()
    
    # 對比快照
    print("\n初始狀態(tài)內(nèi)存占用：")
    display_top_memory(snapshot1, limit=5)
    
    print("\n執(zhí)行后內(nèi)存占用：")
    display_top_memory(snapshot2, limit=5)
    
    # 分析增量
    top_stats = snapshot2.compare_to(snapshot1, 'lineno')
    
    print(f"\n{'='*70}")
    print("內(nèi)存增量分析（Top 10）")
    print(f"{'='*70}")
    
    for stat in top_stats[:10]:
        print(f"\n{stat}")
        if stat.count_diff > 0:
            print(f"  ??  新增對象: {stat.count_diff} 個")
            print(f"  ??  內(nèi)存增加: {stat.size_diff / 1024 / 1024:.2f} MB")
    
    tracemalloc.stop()

# 運行測試
memory_leak_example()

import tracemalloc
from flask import Flask, request
import time

app = Flask(__name__)

# 全局緩存（潛在泄漏點）
request_cache = {}

class MemoryMonitor:
    """內(nèi)存監(jiān)控裝飾器"""
    def __init__(self):
        self.snapshots = []
        tracemalloc.start()
    
    def capture_snapshot(self, label):
        """捕獲內(nèi)存快照"""
        snapshot = tracemalloc.take_snapshot()
        self.snapshots.append((label, snapshot, time.time()))
        
    def analyze_leak(self, threshold_mb=10):
        """分析內(nèi)存泄漏"""
        if len(self.snapshots) < 2:
            print("需要至少兩個快照進行對比")
            return
        
        for i in range(1, len(self.snapshots)):
            label1, snapshot1, time1 = self.snapshots[i-1]
            label2, snapshot2, time2 = self.snapshots[i]
            
            # 計算內(nèi)存增量
            top_stats = snapshot2.compare_to(snapshot1, 'lineno')
            
            total_increase = sum(stat.size_diff for stat in top_stats if stat.size_diff > 0)
            increase_mb = total_increase / 1024 / 1024
            
            print(f"\n{'='*70}")
            print(f"對比: {label1} -> {label2}")
            print(f"時間差: {time2 - time1:.2f}秒")
            print(f"內(nèi)存增加: {increase_mb:.2f} MB")
            print(f"{'='*70}")
            
            if increase_mb > threshold_mb:
                print("??  檢測到可能的內(nèi)存泄漏！")
                print("\n內(nèi)存增長最多的代碼位置：")
                
                for stat in top_stats[:5]:
                    if stat.size_diff > 0:
                        print(f"\n{stat.traceback.format()[0]}")
                        print(f"  增加: {stat.size_diff / 1024 / 1024:.2f} MB")
                        print(f"  新對象: {stat.count_diff} 個")

# 創(chuàng)建監(jiān)控器
monitor = MemoryMonitor()

@app.before_request
def before_request():
    """請求前捕獲快照"""
    request.start_time = time.time()

@app.after_request
def after_request(response):
    """請求后分析內(nèi)存"""
    if hasattr(request, 'start_time'):
        elapsed = time.time() - request.start_time
        if elapsed > 0.1:  # 慢請求
            monitor.capture_snapshot(f"After {request.path}")
    return response

@app.route('/api/process')
def process_data():
    """模擬處理請求（有內(nèi)存泄漏）"""
    request_id = request.args.get('id', 'unknown')
    
    # 泄漏點：緩存永不清理
    large_data = [0] * 100000
    request_cache[request_id] = large_data
    
    return {'status': 'ok', 'cached_requests': len(request_cache)}

@app.route('/api/analyze')
def analyze_memory():
    """觸發(fā)內(nèi)存分析"""
    monitor.analyze_leak(threshold_mb=5)
    return {'status': 'analysis_complete'}

# 運行測試
if __name__ == '__main__':
    # 模擬請求
    with app.test_client() as client:
        monitor.capture_snapshot("Initial")
        
        # 發(fā)送 100 個請求
        for i in range(100):
            client.get(f'/api/process?id={i}')
        
        monitor.capture_snapshot("After 100 requests")
        
        # 再發(fā)送 100 個請求
        for i in range(100, 200):
            client.get(f'/api/process?id={i}')
        
        monitor.capture_snapshot("After 200 requests")
        
        # 分析結(jié)果
        client.get('/api/analyze')

import tracemalloc
import gc

class ObjectTracker:
    """追蹤特定類型對象的內(nèi)存分配"""
    
    @staticmethod
    def track_allocations(target_type, duration_seconds=10):
        """追蹤指定時間內(nèi)的對象分配"""
        tracemalloc.start()
        initial_snapshot = tracemalloc.take_snapshot()
        
        print(f"開始追蹤 {target_type.__name__} 對象，持續(xù) {duration_seconds} 秒...")
        time.sleep(duration_seconds)
        
        final_snapshot = tracemalloc.take_snapshot()
        tracemalloc.stop()
        
        # 分析增量
        top_stats = final_snapshot.compare_to(initial_snapshot, 'lineno')
        
        print(f"\n{target_type.__name__} 對象內(nèi)存分配分析：")
        for stat in top_stats[:10]:
            if target_type.__name__ in str(stat):
                print(f"\n{stat}")
    
    @staticmethod
    def find_object_sources(obj):
        """查找對象的引用來源"""
        print(f"\n{'='*70}")
        print(f"分析對象: {type(obj).__name__} at {hex(id(obj))}")
        print(f"{'='*70}")
        
        # 獲取所有引用該對象的對象
        referrers = gc.get_referrers(obj)
        
        print(f"\n找到 {len(referrers)} 個引用者：")
        for i, ref in enumerate(referrers[:10], 1):
            ref_type = type(ref).__name__
            print(f"\n#{i} 引用者類型: {ref_type}")
            
            if isinstance(ref, dict):
                # 如果是字典，嘗試找到鍵
                for key, value in ref.items():
                    if value is obj:
                        print(f"  字典鍵: {key}")
                        break
            elif isinstance(ref, (list, tuple)):
                print(f"  容器長度: {len(ref)}")
            
            # 顯示引用者的引用者（遞歸查找）
            second_level = gc.get_referrers(ref)
            if second_level:
                print(f"  被 {len(second_level)} 個對象引用")

# 實戰(zhàn)示例
class LeakyCache:
    def __init__(self):
        self.data = {}
    
    def add(self, key, value):
        self.data[key] = value

# 測試
cache = LeakyCache()
for i in range(1000):
    cache.add(f"key_{i}", [0] * 10000)

# 追蹤泄漏源
ObjectTracker.find_object_sources(cache.data)

# 安裝
pip install objgraph

# 生成圖譜需要 Graphviz
# Ubuntu/Debian
sudo apt-get install graphviz

# macOS
brew install graphviz

# Windows
# 從 https://graphviz.org/download/ 下載安裝

import objgraph
import gc

# 基礎(chǔ)統(tǒng)計
def analyze_object_types():
    """分析當前內(nèi)存中的對象類型"""
    print("\n內(nèi)存中最多的對象類型（Top 20）：")
    objgraph.show_most_common_types(limit=20)

# 增長分析
def track_object_growth():
    """追蹤對象數(shù)量增長"""
    # 第一次統(tǒng)計
    gc.collect()
    objgraph.show_growth(limit=10)
    
    # 創(chuàng)建一些對象
    leaked_list = []
    for i in range(10000):
        leaked_list.append({'data': [0] * 100})
    
    # 第二次統(tǒng)計
    print("\n執(zhí)行操作后的對象增長：")
    objgraph.show_growth(limit=10)

# 運行分析
analyze_object_types()
track_object_growth()

import objgraph
import os

class Node:
    """鏈表節(jié)點（可能產(chǎn)生循環(huán)引用）"""
    def __init__(self, value):
        self.value = value
        self.next = None
        self.prev = None

class CircularList:
    """循環(huán)鏈表（演示內(nèi)存泄漏）"""
    def __init__(self):
        self.head = None
        self.size = 0
    
    def add(self, value):
        new_node = Node(value)
        if not self.head:
            self.head = new_node
            new_node.next = new_node
            new_node.prev = new_node
        else:
            tail = self.head.prev
            tail.next = new_node
            new_node.prev = tail
            new_node.next = self.head
            self.head.prev = new_node
        
        self.size += 1

# 創(chuàng)建循環(huán)引用
def create_circular_references():
    """創(chuàng)建包含循環(huán)引用的對象"""
    lists = []
    for i in range(10):
        circular_list = CircularList()
        for j in range(100):
            circular_list.add(f"data_{i}_{j}")
        lists.append(circular_list)
    
    return lists

# 可視化分析
def visualize_references():
    """生成對象引用關(guān)系圖"""
    # 創(chuàng)建對象
    leaked_lists = create_circular_references()
    
    # 分析第一個列表
    target = leaked_lists[0]
    
    print("\n生成對象引用關(guān)系圖...")
    
    # 生成反向引用鏈（是什么在引用這個對象）
    output_file = '/home/claude/backrefs.png'
    objgraph.show_backrefs(
        [target],
        max_depth=3,
        filename=output_file,
        refcounts=True
    )
    print(f"反向引用圖已保存: {output_file}")
    
    # 生成前向引用鏈（這個對象引用了什么）
    output_file = '/home/claude/refs.png'
    objgraph.show_refs(
        [target.head],
        max_depth=3,
        filename=output_file,
        refcounts=True
    )
    print(f"前向引用圖已保存: {output_file}")
    
    return leaked_lists

# 運行可視化
leaked = visualize_references()

# 查看引用鏈
print("\n詳細引用鏈分析：")
objgraph.show_chain(
    objgraph.find_backref_chain(
        leaked[0],
        objgraph.is_proper_module
    ),
    filename='/home/claude/chain.png'
)

import objgraph
import tracemalloc
import gc
from functools import wraps

class MemoryLeakDetector:
    """內(nèi)存泄漏檢測器（生產(chǎn)環(huán)境友好）"""
    
    def __init__(self, threshold_mb=50):
        self.threshold_mb = threshold_mb
        self.baseline = None
        self.snapshots = []
        
    def start_monitoring(self):
        """開始監(jiān)控"""
        gc.collect()
        tracemalloc.start()
        self.baseline = tracemalloc.take_snapshot()
        print("? 內(nèi)存監(jiān)控已啟動")
    
    def check_memory(self, label="checkpoint"):
        """檢查內(nèi)存狀態(tài)"""
        if not self.baseline:
            print("??  請先調(diào)用 start_monitoring()")
            return
        
        gc.collect()
        current = tracemalloc.take_snapshot()
        self.snapshots.append((label, current))
        
        # 計算增量
        stats = current.compare_to(self.baseline, 'lineno')
        total_increase = sum(s.size_diff for s in stats if s.size_diff > 0)
        increase_mb = total_increase / 1024 / 1024
        
        print(f"\n{'='*70}")
        print(f"檢查點: {label}")
        print(f"內(nèi)存增長: {increase_mb:.2f} MB")
        
        if increase_mb > self.threshold_mb:
            print("?? 檢測到內(nèi)存泄漏！")
            self._analyze_leak(stats)
        else:
            print("? 內(nèi)存使用正常")
        print(f"{'='*70}")
    
    def _analyze_leak(self, stats):
        """詳細分析泄漏"""
        print("\n內(nèi)存增長最多的位置（Top 10）：")
        for i, stat in enumerate(stats[:10], 1):
            if stat.size_diff > 0:
                print(f"\n#{i}: {stat.traceback.format()[0]}")
                print(f"  增長: {stat.size_diff / 1024 / 1024:.2f} MB")
                print(f"  對象: +{stat.count_diff}")
        
        # 使用 objgraph 分析對象類型
        print("\n對象類型增長分析：")
        objgraph.show_growth(limit=10)
    
    def generate_report(self, output_dir='/home/claude'):
        """生成完整報告"""
        print(f"\n生成內(nèi)存泄漏報告...")
        
        # 1. 對象類型統(tǒng)計
        print("\n1. 當前內(nèi)存對象類型分布：")
        objgraph.show_most_common_types(limit=15)
        
        # 2. 查找潛在泄漏對象
        print("\n2. 查找可疑對象...")
        suspicious_types = ['dict', 'list', 'tuple', 'set']
        
        for obj_type in suspicious_types:
            objects = objgraph.by_type(obj_type)
            if len(objects) > 10000:
                print(f"\n??  {obj_type} 對象數(shù)量異常: {len(objects)}")
                
                # 隨機采樣分析
                sample = objects[0] if objects else None
                if sample:
                    output_file = os.path.join(output_dir, f'{obj_type}_refs.png')
                    objgraph.show_refs(
                        [sample],
                        filename=output_file,
                        max_depth=2
                    )
                    print(f"  引用圖已保存: {output_file}")
        
        # 3. tracemalloc 詳細報告
        if self.snapshots:
            latest_label, latest_snapshot = self.snapshots[-1]
            print(f"\n3. 最新快照分析 ({latest_label})：")
            
            top_stats = latest_snapshot.statistics('lineno')
            print("\n內(nèi)存占用 Top 10:")
            for i, stat in enumerate(top_stats[:10], 1):
                frame = stat.traceback[0]
                print(f"\n#{i}: {frame.filename}:{frame.lineno}")
                print(f"  大小: {stat.size / 1024 / 1024:.2f} MB")
                print(f"  對象數(shù): {stat.count}")

# 裝飾器：自動檢測函數(shù)內(nèi)存泄漏
def detect_leak(detector):
    """裝飾器：自動檢測函數(shù)執(zhí)行后的內(nèi)存變化"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            gc.collect()
            before = tracemalloc.take_snapshot()
            
            result = func(*args, **kwargs)
            
            gc.collect()
            after = tracemalloc.take_snapshot()
            
            stats = after.compare_to(before, 'lineno')
            total_increase = sum(s.size_diff for s in stats if s.size_diff > 0)
            increase_mb = total_increase / 1024 / 1024
            
            if increase_mb > 1:  # 閾值 1MB
                print(f"\n??  {func.__name__} 可能存在內(nèi)存泄漏")
                print(f"   內(nèi)存增長: {increase_mb:.2f} MB")
                
                for stat in stats[:3]:
                    if stat.size_diff > 0:
                        print(f"   {stat}")
            
            return result
        return wrapper
    return decorator

# 使用示例
detector = MemoryLeakDetector(threshold_mb=10)
detector.start_monitoring()

@detect_leak(detector)
def process_large_dataset():
    """模擬數(shù)據(jù)處理（有泄漏）"""
    cache = {}
    for i in range(50000):
        cache[f"key_{i}"] = [0] * 1000  # 泄漏點
    return len(cache)

# 測試
result = process_large_dataset()
detector.check_memory("After processing")
detector.generate_report()

import tracemalloc
import objgraph
import gc
import psutil
import os

class MemoryDebugger:
    """內(nèi)存調(diào)試完整工作流"""
    
    @staticmethod
    def step1_confirm_leak():
        """步驟1：確認是否真的有內(nèi)存泄漏"""
        print("="*70)
        print("步驟 1: 確認內(nèi)存泄漏")
        print("="*70)
        
        process = psutil.Process(os.getpid())
        
        baseline = process.memory_info().rss / 1024 / 1024
        print(f"基線內(nèi)存: {baseline:.2f} MB")
        
        # 模擬工作負載
        for iteration in range(5):
            # 執(zhí)行業(yè)務邏輯
            _ = [0] * 1000000
            gc.collect()
            
            current = process.memory_info().rss / 1024 / 1024
            increase = current - baseline
            
            print(f"迭代 {iteration + 1}: {current:.2f} MB (+{increase:.2f} MB)")
            
            if increase > 100:
                print("??  確認內(nèi)存持續(xù)增長，可能存在泄漏！")
                return True
        
        print("? 內(nèi)存使用正常")
        return False
    
    @staticmethod
    def step2_locate_source():
        """步驟2：使用 tracemalloc 定位泄漏源"""
        print("\n" + "="*70)
        print("步驟 2: 定位泄漏源")
        print("="*70)
        
        tracemalloc.start()
        snapshot1 = tracemalloc.take_snapshot()
        
        # 執(zhí)行可疑代碼
        leaked_data = []
        for i in range(10000):
            leaked_data.append([0] * 1000)
        
        snapshot2 = tracemalloc.take_snapshot()
        
        top_stats = snapshot2.compare_to(snapshot1, 'lineno')
        
        print("\n內(nèi)存增長最多的代碼位置:")
        for stat in top_stats[:5]:
            if stat.size_diff > 0:
                print(f"\n{stat.traceback.format()[0]}")
                print(f"增長: {stat.size_diff / 1024 / 1024:.2f} MB")
        
        tracemalloc.stop()
    
    @staticmethod
    def step3_analyze_objects():
        """步驟3：使用 objgraph 分析對象關(guān)系"""
        print("\n" + "="*70)
        print("步驟 3: 分析對象關(guān)系")
        print("="*70)
        
        # 查看對象增長
        gc.collect()
        print("\n初始對象統(tǒng)計:")
        objgraph.show_growth(limit=10)
        
        # 創(chuàng)建泄漏
        global leaked_cache
        leaked_cache = {}
        for i in range(5000):
            leaked_cache[i] = [0] * 1000
        
        print("\n操作后對象增長:")
        objgraph.show_growth(limit=10)
        
        # 生成引用圖
        if leaked_cache:
            sample_obj = list(leaked_cache.values())[0]
            objgraph.show_backrefs(
                [sample_obj],
                filename='/home/claude/leak_backrefs.png',
                max_depth=3
            )
            print("\n引用圖已生成: /home/claude/leak_backrefs.png")
    
    @staticmethod
    def step4_verify_fix():
        """步驟4：驗證修復效果"""
        print("\n" + "="*70)
        print("步驟 4: 驗證修復")
        print("="*70)
        
        tracemalloc.start()
        before = tracemalloc.take_snapshot()
        
        # 修復后的代碼（使用弱引用或限制緩存大小）
        from collections import OrderedDict
        
        class LRUCache:
            def __init__(self, max_size=1000):
                self.cache = OrderedDict()
                self.max_size = max_size
            
            def set(self, key, value):
                if key in self.cache:
                    self.cache.move_to_end(key)
                self.cache[key] = value
                if len(self.cache) > self.max_size:
                    self.cache.popitem(last=False)
        
        cache = LRUCache(max_size=1000)
        for i in range(10000):
            cache.set(i, [0] * 1000)
        
        after = tracemalloc.take_snapshot()
        stats = after.compare_to(before, 'lineno')
        
        total_increase = sum(s.size_diff for s in stats if s.size_diff > 0)
        print(f"\n修復后內(nèi)存增長: {total_increase / 1024 / 1024:.2f} MB")
        
        if total_increase / 1024 / 1024 < 10:
            print("? 修復有效，內(nèi)存控制在合理范圍")
        else:
            print("??  仍需進一步優(yōu)化")
        
        tracemalloc.stop()

# 執(zhí)行完整診斷流程
if __name__ == '__main__':
    debugger = MemoryDebugger()
    
    if debugger.step1_confirm_leak():
        debugger.step2_locate_source()
        debugger.step3_analyze_objects()
        debugger.step4_verify_fix()

import tracemalloc
import threading
import time
from datetime import datetime

class ProductionMemoryMonitor:
    """生產(chǎn)環(huán)境內(nèi)存監(jiān)控（低開銷）"""
    
    def __init__(self, check_interval=300, alert_threshold_mb=500):
        self.check_interval = check_interval
        self.alert_threshold_mb = alert_threshold_mb
        self.running = False
        self.thread = None
        
    def start(self):
        """啟動監(jiān)控線程"""
        if self.running:
            return
        
        self.running = True
        tracemalloc.start()
        
        self.thread = threading.Thread(target=self._monitor_loop, daemon=True)
        self.thread.start()
        print(f"? 內(nèi)存監(jiān)控已啟動（每 {self.check_interval} 秒檢查一次）")
    
    def stop(self):
        """停止監(jiān)控"""
        self.running = False
        if self.thread:
            self.thread.join()
        tracemalloc.stop()
        print("?  內(nèi)存監(jiān)控已停止")
    
    def _monitor_loop(self):
        """監(jiān)控循環(huán)"""
        baseline = None
        
        while self.running:
            try:
                snapshot = tracemalloc.take_snapshot()
                
                if baseline is None:
                    baseline = snapshot
                else:
                    self._check_memory(baseline, snapshot)
                
                time.sleep(self.check_interval)
                
            except Exception as e:
                print(f"監(jiān)控出錯: {e}")
    
    def _check_memory(self, baseline, current):
        """檢查內(nèi)存狀態(tài)"""
        stats = current.compare_to(baseline, 'lineno')
        total_increase = sum(s.size_diff for s in stats if s.size_diff > 0)
        increase_mb = total_increase / 1024 / 1024
        
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        
        if increase_mb > self.alert_threshold_mb:
            print(f"\n?? [{timestamp}] 內(nèi)存告警！")
            print(f"   增長: {increase_mb:.2f} MB")
            print(f"   Top 3 增長位置:")
            
            for i, stat in enumerate(stats[:3], 1):
                if stat.size_diff > 0:
                    print(f"   #{i}: {stat.traceback.format()[0]}")
                    print(f"       +{stat.size_diff / 1024 / 1024:.2f} MB")
            
            # 可以在這里發(fā)送告警郵件或消息
        else:
            print(f"? [{timestamp}] 內(nèi)存正常 (+{increase_mb:.2f} MB)")

# 使用示例
monitor = ProductionMemoryMonitor(check_interval=10, alert_threshold_mb=50)
monitor.start()

# 模擬應用運行
try:
    leaked = []
    for i in range(100):
        leaked.append([0] * 100000)
        time.sleep(1)
except KeyboardInterrupt:
    pass
finally:
    monitor.stop()

# 1. 使用上下文管理器
with open('file.txt') as f:
    data = f.read()

# 2. 限制緩存大小
from functools import lru_cache

@lru_cache(maxsize=1000)
def expensive_function(arg):
    return arg ** 2

# 3. 使用弱引用
import weakref

class Cache:
    def __init__(self):
        self._cache = weakref.WeakValueDictionary()

# 4. 定期清理
def cleanup_old_data(cache, max_age_seconds=3600):
    now = time.time()
    to_delete = [
        k for k, v in cache.items()
        if now - v['timestamp'] > max_age_seconds
    ]
    for k in to_delete:
        del cache[k]

# 5. 使用生成器處理大數(shù)據(jù)
def process_large_file(filename):
    with open(filename) as f:
        for line in f:  # 逐行處理，不加載整個文件
            yield process_line(line)