# 1. 循环引用未被GC回收
class Node:
    def __init__(self):
        self.parent = None
        self.children = []
    
    def add_child(self, child):
        child.parent = self
        self.children.append(child)

# 2. 全局变量持续增长
GLOBAL_CACHE = {}

def leak_function(key, data):
    GLOBAL_CACHE[key] = data  # 无清理机制

# 3. 资源未正确释放
def file_leak():
    f = open('file.txt')
    # 未调用 f.close()
    return f.read()

# 4. 事件监听器未注销
class EventManager:
    def __init__(self):
        self.listeners = []
    
    def add_listener(self, callback):
        self.listeners.append(callback)  # 未提供移除机制

import psutil
import time
import threading

class MemoryMonitor:
    def __init__(self, interval=1):
        self.interval = interval
        self.running = False
        self.memory_data = []
    
    def start(self):
        self.running = True
        self.thread = threading.Thread(target=self._monitor)
        self.thread.start()
    
    def stop(self):
        self.running = False
        self.thread.join()
    
    def _monitor(self):
        process = psutil.Process()
        while self.running:
            memory_info = process.memory_info()
            self.memory_data.append({
                'timestamp': time.time(),
                'rss': memory_info.rss / 1024 / 1024,  # MB
                'vms': memory_info.vms / 1024 / 1024   # MB
            })
            time.sleep(self.interval)
    
    def get_memory_trend(self):
        if len(self.memory_data) < 2:
            return 0
        return (self.memory_data[-1]['rss'] - self.memory_data[0]['rss']) / len(self.memory_data)

# 使用示例
monitor = MemoryMonitor()
monitor.start()

# 运行可能有内存泄漏的代码
for i in range(1000):
    # 你的代码
    pass

monitor.stop()
trend = monitor.get_memory_trend()
if trend > 0.1:  # 每次迭代增长0.1MB
    print(f"检测到潜在内存泄漏，增长趋势: {trend:.2f} MB/iteration")

import gc
from collections import defaultdict

def get_object_stats():
    """获取当前Python对象统计信息"""
    stats = defaultdict(int)
    for obj in gc.get_objects():
        obj_type = type(obj).__name__
        stats[obj_type] += 1
    return dict(stats)

def detect_object_leak(iterations=100):
    """检测对象数量异常增长"""
    initial_stats = get_object_stats()
    
    # 运行测试代码
    for i in range(iterations):
        # 你的代码
        pass
    
    final_stats = get_object_stats()
    
    # 分析增长
    growth = {}
    for obj_type, final_count in final_stats.items():
        initial_count = initial_stats.get(obj_type, 0)
        if final_count > initial_count:
            growth[obj_type] = final_count - initial_count
    
    # 按增长量排序
    return sorted(growth.items(), key=lambda x: x[1], reverse=True)

# 使用示例
leaked_objects = detect_object_leak()
for obj_type, count in leaked_objects[:10]:
    print(f"{obj_type}: +{count}")

import tracemalloc
import time

class TraceMallocProfiler:
    def __init__(self):
        self.snapshots = []
    
    def start(self):
        tracemalloc.start(10)  # 保留10层调用栈
        self.take_snapshot("initial")
    
    def take_snapshot(self, label):
        snapshot = tracemalloc.take_snapshot()
        self.snapshots.append((label, snapshot, time.time()))
    
    def analyze_growth(self, top_n=10):
        if len(self.snapshots) < 2:
            return []
        
        _, initial_snapshot, _ = self.snapshots[0]
        _, final_snapshot, _ = self.snapshots[-1]
        
        top_stats = final_snapshot.compare_to(initial_snapshot, 'lineno')
        
        results = []
        for stat in top_stats[:top_n]:
            results.append({
                'size_diff': stat.size_diff / 1024 / 1024,  # MB
                'count_diff': stat.count_diff,
                'filename': stat.traceback.format()[0],
                'traceback': stat.traceback.format()
            })
        
        return results
    
    def stop(self):
        tracemalloc.stop()

# 使用示例
profiler = TraceMallocProfiler()
profiler.start()

# 运行可能有内存泄漏的代码
for i in range(100):
    # 你的代码
    if i % 20 == 0:
        profiler.take_snapshot(f"iteration_{i}")

results = profiler.analyze_growth()
for result in results:
    print(f"内存增长: {result['size_diff']:.2f} MB")
    print(f"对象增长: {result['count_diff']}")
    print(f"位置: {result['filename']}")
    print("调用栈:")
    for line in result['traceback']:
        print(f"  {line}")
    print("-" * 50)

profiler.stop()

import objgraph
import gc

class ObjectGraphAnalyzer:
    def __init__(self):
        self.baseline_counts = {}
    
    def set_baseline(self):
        """设置基线对象计数"""
        self.baseline_counts = objgraph.typestats()
    
    def find_growing_objects(self, top_n=10):
        """找到数量增长的对象类型"""
        current_counts = objgraph.typestats()
        growth = {}
        
        for obj_type, current_count in current_counts.items():
            baseline_count = self.baseline_counts.get(obj_type, 0)
            if current_count > baseline_count:
                growth[obj_type] = current_count - baseline_count
        
        return sorted(growth.items(), key=lambda x: x[1], reverse=True)[:top_n]
    
    def analyze_references(self, obj_type, max_objects=3):
        """分析特定类型对象的引用关系"""
        objects = objgraph.by_type(obj_type)
        if not objects:
            return
        
        for i, obj in enumerate(objects[:max_objects]):
            filename = f"refs_{obj_type}_{i}.png"
            print(f"生成引用图: {filename}")
            objgraph.show_refs([obj], filename=filename, max_depth=5)
    
    def find_uncollectable_cycles(self):
        """查找不可回收的循环引用"""
        gc.collect()
        if gc.garbage:
            print(f"发现 {len(gc.garbage)} 个不可回收对象")
            for i, obj in enumerate(gc.garbage[:5]):
                filename = f"cycle_{i}.png"
                objgraph.show_refs([obj], filename=filename)
                objgraph.show_backrefs([obj], filename=f"back_{filename}")

# 使用示例
analyzer = ObjectGraphAnalyzer()
analyzer.set_baseline()

# 运行可能有内存泄漏的代码
for i in range(100):
    # 你的代码
    pass

growing_objects = analyzer.find_growing_objects()
for obj_type, growth in growing_objects:
    print(f"{obj_type}: +{growth}")
    analyzer.analyze_references(obj_type)

analyzer.find_uncollectable_cycles()

import sys
import gc
from collections import defaultdict

class MemorySnapshotAnalyzer:
    def __init__(self):
        self.snapshots = []
    
    def take_snapshot(self, label):
        """获取内存快照"""
        gc.collect()  # 强制垃圾回收
        
        objects_by_type = defaultdict(list)
        for obj in gc.get_objects():
            obj_type = type(obj).__name__
            objects_by_type[obj_type].append(obj)
        
        snapshot = {
            'label': label,
            'total_objects': len(gc.get_objects()),
            'objects_by_type': dict(objects_by_type),
            'ref_count': sys.gettotalrefcount() if hasattr(sys, 'gettotalrefcount') else 0
        }
        
        self.snapshots.append(snapshot)
        return snapshot
    
    def compare_snapshots(self, snapshot1_idx=0, snapshot2_idx=-1):
        """比较两个快照的差异"""
        if len(self.snapshots) < 2:
            return {}
        
        snap1 = self.snapshots[snapshot1_idx]
        snap2 = self.snapshots[snapshot2_idx]
        
        diff = {
            'total_objects_diff': snap2['total_objects'] - snap1['total_objects'],
            'ref_count_diff': snap2['ref_count'] - snap1['ref_count'],
            'object_type_diffs': {}
        }
        
        # 分析每种对象类型的变化
        all_types = set(snap1['objects_by_type'].keys()) | set(snap2['objects_by_type'].keys())
        
        for obj_type in all_types:
            count1 = len(snap1['objects_by_type'].get(obj_type, []))
            count2 = len(snap2['objects_by_type'].get(obj_type, []))
            if count2 != count1:
                diff['object_type_diffs'][obj_type] = {
                    'before': count1,
                    'after': count2,
                    'diff': count2 - count1
                }
        
        return diff
    
    def find_leaked_objects(self, obj_type, snapshot1_idx=0, snapshot2_idx=-1):
        """找到特定类型的泄漏对象"""
        snap1 = self.snapshots[snapshot1_idx]
        snap2 = self.snapshots[snapshot2_idx]
        
        objects1 = set(id(obj) for obj in snap1['objects_by_type'].get(obj_type, []))
        objects2 = snap2['objects_by_type'].get(obj_type, [])
        
        # 找到新增的对象
        leaked_objects = [obj for obj in objects2 if id(obj) not in objects1]
        return leaked_objects

# 使用示例
analyzer = MemorySnapshotAnalyzer()
analyzer.take_snapshot("baseline")

# 运行可能有内存泄漏的代码
for i in range(100):
    # 你的代码
    if i % 25 == 0:
        analyzer.take_snapshot(f"iteration_{i}")

# 分析结果
diff = analyzer.compare_snapshots()
print(f"总对象数变化: {diff['total_objects_diff']}")
print(f"引用计数变化: {diff['ref_count_diff']}")

print("\n对象类型变化:")
for obj_type, change in sorted(diff['object_type_diffs'].items(), 
                               key=lambda x: x[1]['diff'], reverse=True)[:10]:
    print(f"{obj_type}: {change['before']} -> {change['after']} ({change['diff']:+d})")

# 分析特定类型的泄漏对象
if 'dict' in diff['object_type_diffs']:
    leaked_dicts = analyzer.find_leaked_objects('dict')
    print(f"\n发现 {len(leaked_dicts)} 个泄漏的dict对象")

import weakref
from collections import defaultdict

class WeakRefCache:
    """使用弱引用避免循环引用"""
    def __init__(self):
        self._cache = weakref.WeakValueDictionary()
        self._callbacks = defaultdict(list)
    
    def get_or_create(self, key, factory):
        obj = self._cache.get(key)
        if obj is None:
            obj = factory()
            self._cache[key] = obj
            # 注册清理回调
            weakref.ref(obj, lambda ref: self._cleanup(key))
        return obj
    
    def _cleanup(self, key):
        """对象被回收时的清理逻辑"""
        callbacks = self._callbacks.get(key, [])
        for callback in callbacks:
            callback()
        del self._callbacks[key]

# 使用示例
cache = WeakRefCache()

class ExpensiveObject:
    def __init__(self, data):
        self.data = data

def create_expensive_object():
    return ExpensiveObject("some data")

# 对象会在不再被引用时自动清理
obj = cache.get_or_create("key1", create_expensive_object)

import contextlib
from threading import RLock

class ResourceManager:
    """资源管理器确保资源正确释放"""
    def __init__(self):
        self._resources = {}
        self._lock = RLock()
    
    @contextlib.contextmanager
    def acquire_resource(self, resource_id, factory):
        resource = None
        try:
            with self._lock:
                if resource_id not in self._resources:
                    self._resources[resource_id] = factory()
                resource = self._resources[resource_id]
            yield resource
        finally:
            # 确保资源被正确释放
            if resource and hasattr(resource, 'close'):
                resource.close()
            with self._lock:
                if resource_id in self._resources:
                    del self._resources[resource_id]

# 使用示例
manager = ResourceManager()

class DatabaseConnection:
    def __init__(self, url):
        self.url = url
        self.connected = True
        print(f"连接到数据库: {url}")
    
    def close(self):
        if self.connected:
            self.connected = False
            print(f"关闭数据库连接: {self.url}")

def create_db_connection():
    return DatabaseConnection("postgresql://localhost/mydb")

# 自动管理资源生命周期
with manager.acquire_resource("db", create_db_connection) as db:
    # 使用数据库连接
    pass
# 连接自动关闭

import threading
import time
from queue import Queue, Empty

class ObjectPool:
    """对象池避免频繁创建销毁对象"""
    def __init__(self, factory, max_size=10, max_age=300):
        self.factory = factory
        self.max_size = max_size
        self.max_age = max_age
        self._pool = Queue(maxsize=max_size)
        self._created_objects = {}
        self._lock = threading.Lock()
    
    @contextlib.contextmanager
    def get_object(self):
        obj = self._acquire()
        try:
            yield obj
        finally:
            self._release(obj)
    
    def _acquire(self):
        try:
            # 尝试从池中获取对象
            obj, created_time = self._pool.get_nowait()
            if time.time() - created_time > self.max_age:
                # 对象太老，创建新的
                if hasattr(obj, 'close'):
                    obj.close()
                obj = self.factory()
                created_time = time.time()
        except Empty:
            # 池为空，创建新对象
            obj = self.factory()
            created_time = time.time()
        
        with self._lock:
            self._created_objects[id(obj)] = created_time
        
        return obj
    
    def _release(self, obj):
        with self._lock:
            if id(obj) in self._created_objects:
                created_time = self._created_objects[id(obj)]
                try:
                    self._pool.put_nowait((obj, created_time))
                except:
                    # 池已满，销毁对象
                    if hasattr(obj, 'close'):
                        obj.close()
                    del self._created_objects[id(obj)]

# 使用示例
def create_connection():
    return DatabaseConnection("postgresql://localhost/mydb")

pool = ObjectPool(create_connection, max_size=5)

with pool.get_object() as conn:
    # 使用连接
    pass

class OptimizedClass:
    """使用__slots__减少内存使用"""
    __slots__ = ['x', 'y', 'z']
    
    def __init__(self, x, y, z):
        self.x = x
        self.y = y
        self.z = z

class RegularClass:
    def __init__(self, x, y, z):
        self.x = x
        self.y = y
        self.z = z

# 内存使用对比
import sys

def compare_memory_usage():
    optimized = OptimizedClass(1, 2, 3)
    regular = RegularClass(1, 2, 3)
    
    print(f"优化类内存使用: {sys.getsizeof(optimized)} bytes")
    print(f"普通类内存使用: {sys.getsizeof(regular)} bytes")
    if hasattr(regular, '__dict__'):
        print(f"普通类__dict__: {sys.getsizeof(regular.__dict__)} bytes")

compare_memory_usage()

def memory_efficient_processing(data_source):
    """使用生成器避免加载大量数据到内存"""
    def process_batch(batch):
        for item in batch:
            # 处理每个项目
            yield transform(item)
    
    def transform(item):
        # 数据转换逻辑
        return item * 2
    
    batch_size = 1000
    batch = []
    
    for item in data_source:
        batch.append(item)
        if len(batch) >= batch_size:
            yield from process_batch(batch)
            batch.clear()  # 清空批次释放内存
    
    # 处理最后一个批次
    if batch:
        yield from process_batch(batch)

# 内存友好的大数据处理
def process_large_dataset(filename):
    def read_data():
        with open(filename, 'r') as f:
            for line in f:
                yield line.strip()
    
    # 使用生成器链避免内存占用
    processed_data = memory_efficient_processing(read_data())
    
    for item in processed_data:
        # 处理每个项目
        pass

import unittest
import psutil
import gc
import time
from functools import wraps

class MemoryLeakTestCase(unittest.TestCase):
    """内存泄漏测试基类"""
    
    def setUp(self):
        """测试前准备"""
        gc.collect()
        self.initial_memory = self._get_memory_usage()
        self.initial_objects = len(gc.get_objects())
    
    def tearDown(self):
        """测试后清理"""
        gc.collect()
        final_memory = self._get_memory_usage()
        final_objects = len(gc.get_objects())
        
        memory_growth = final_memory - self.initial_memory
        object_growth = final_objects - self.initial_objects
        
        # 内存增长超过阈值时报告
        if memory_growth > 10:  # 10MB
            self.fail(f"内存增长过多: {memory_growth:.2f} MB")
        
        if object_growth > 1000:  # 1000个对象
            self.fail(f"对象数量增长过多: {object_growth}")
    
    def _get_memory_usage(self):
        """获取当前内存使用量(MB)"""
        process = psutil.Process()
        return process.memory_info().rss / 1024 / 1024

def memory_leak_test(iterations=100, max_memory_growth=5):
    """内存泄漏测试装饰器"""
    def decorator(test_func):
        @wraps(test_func)
        def wrapper(*args, **kwargs):
            gc.collect()
            initial_memory = psutil.Process().memory_info().rss / 1024 / 1024
            
            for i in range(iterations):
                test_func(*args, **kwargs)
                if i % 10 == 0:
                    gc.collect()
            
            final_memory = psutil.Process().memory_info().rss / 1024 / 1024
            memory_growth = final_memory - initial_memory
            
            if memory_growth > max_memory_growth:
                raise AssertionError(f"内存泄漏检测失败: 增长 {memory_growth:.2f} MB")
            
            return True
        return wrapper
    return decorator

class TestMemoryLeaks(MemoryLeakTestCase):
    
    @memory_leak_test(iterations=1000, max_memory_growth=2)
    def test_cache_function(self):
        """测试缓存函数是否有内存泄漏"""
        cache = {}
        
        def cached_function(x):
            if x not in cache:
                cache[x] = x * x
            return cache[x]
        
        # 测试缓存函数
        result = cached_function(42)
        self.assertEqual(result, 1764)
    
    def test_circular_reference(self):
        """测试循环引用是否正确处理"""
        
        class Node:
            def __init__(self, value):
                self.value = value
                self.parent = None
                self.children = []
        
        # 创建循环引用
        parent = Node("parent")
        child = Node("child")
        parent.children.append(child)
        child.parent = parent
        
        # 模拟使用后删除引用
        del parent, child
        
        # 验证垃圾回收
        gc.collect()

if __name__ == '__main__':
    unittest.main()

import time
import json
import logging
from datetime import datetime
from threading import Thread
import psutil

class MemoryMonitoringSystem:
    """内存监控系统"""
    
    def __init__(self, check_interval=60, alert_threshold=1000):
        self.check_interval = check_interval  # 检查间隔(秒)
        self.alert_threshold = alert_threshold  # 警报阈值(MB)
        self.running = False
        self.memory_history = []
        self.logger = self._setup_logger()
    
    def _setup_logger(self):
        logger = logging.getLogger('memory_monitor')
        logger.setLevel(logging.INFO)
        
        handler = logging.FileHandler('memory_monitor.log')
        formatter = logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)
        logger.addHandler(handler)
        
        return logger
    
    def start(self):
        """启动监控"""
        self.running = True
        self.monitor_thread = Thread(target=self._monitor_loop)
        self.monitor_thread.start()
        self.logger.info("内存监控系统启动")
    
    def stop(self):
        """停止监控"""
        self.running = False
        if hasattr(self, 'monitor_thread'):
            self.monitor_thread.join()
        self.logger.info("内存监控系统停止")
    
    def _monitor_loop(self):
        """监控循环"""
        while self.running:
            try:
                memory_info = self._collect_memory_info()
                self.memory_history.append(memory_info)
                
                # 保留最近24小时的数据
                cutoff_time = time.time() - 24 * 3600
                self.memory_history = [
                    info for info in self.memory_history 
                    if info['timestamp'] > cutoff_time
                ]
                
                # 检查是否需要报警
                self._check_alerts(memory_info)
                
                # 生成报告
                if len(self.memory_history) % 60 == 0:  # 每小时生成一次报告
                    self._generate_report()
                
            except Exception as e:
                self.logger.error(f"监控过程中发生错误: {e}")
            
            time.sleep(self.check_interval)
    
    def _collect_memory_info(self):
        """收集内存信息"""
        process = psutil.Process()
        memory_info = process.memory_info()
        
        return {
            'timestamp': time.time(),
            'rss': memory_info.rss / 1024 / 1024,  # MB
            'vms': memory_info.vms / 1024 / 1024,  # MB
            'percent': process.memory_percent(),
            'available': psutil.virtual_memory().available / 1024 / 1024,  # MB
            'objects_count': len(gc.get_objects()) if 'gc' in globals() else 0
        }
    
    def _check_alerts(self, memory_info):
        """检查警报条件"""
        if memory_info['rss'] > self.alert_threshold:
            self.logger.warning(
                f"内存使用超过阈值: {memory_info['rss']:.2f} MB > {self.alert_threshold} MB"
            )
        
        # 检查内存增长趋势
        if len(self.memory_history) >= 10:
            recent_memories = [info['rss'] for info in self.memory_history[-10:]]
            if self._is_growing_trend(recent_memories):
                self.logger.warning("检测到内存持续增长趋势")
    
    def _is_growing_trend(self, values, threshold=0.05):
        """检测是否有增长趋势"""
        if len(values) < 3:
            return False
        
        # 计算线性回归斜率
        n = len(values)
        sum_x = sum(range(n))
        sum_y = sum(values)
        sum_xy = sum(i * values[i] for i in range(n))
        sum_x2 = sum(i * i for i in range(n))
        
        slope = (n * sum_xy - sum_x * sum_y) / (n * sum_x2 - sum_x * sum_x)
        
        return slope > threshold
    
    def _generate_report(self):
        """生成内存使用报告"""
        if not self.memory_history:
            return
        
        recent_data = self.memory_history[-60:]  # 最近一小时的数据
        
        report = {
            'timestamp': datetime.now().isoformat(),
            'period': 'last_hour',
            'stats': {
                'min_memory': min(info['rss'] for info in recent_data),
                'max_memory': max(info['rss'] for info in recent_data),
                'avg_memory': sum(info['rss'] for info in recent_data) / len(recent_data),
                'current_memory': recent_data[-1]['rss'],
                'total_objects': recent_data[-1]['objects_count']
            }
        }
        
        self.logger.info(f"内存使用报告: {json.dumps(report, indent=2)}")

# 使用示例
monitor = MemoryMonitoringSystem(check_interval=30, alert_threshold=500)
monitor.start()

try:
    # 运行你的应用程序
    time.sleep(3600)  # 运行1小时
finally:
    monitor.stop()

class MemoryLeakCodeReviewChecklist:
    """内存泄漏代码审查检查清单"""
    
    def __init__(self):
        self.checklist = {
            'resource_management': [
                '文件操作是否使用with语句或正确关闭',
                '数据库连接是否正确释放',
                '网络连接是否正确关闭',
                '线程和进程是否正确终止'
            ],
            'circular_references': [
                '是否存在对象间的循环引用',
                '事件监听器是否有注销机制',
                '缓存是否有清理策略',
                '全局变量是否持续增长'
            ],
            'memory_optimization': [
                '是否使用了__slots__优化小对象',
                '大数据处理是否使用生成器',
                '是否有不必要的数据复制',
                '缓存大小是否有限制'
            ],
            'gc_considerations': [
                '是否正确处理弱引用',
                '是否避免在__del__中进行复杂操作',
                '循环引用是否可以被GC回收',
                '是否有人为禁用GC的代码'
            ]
        }
    
    def review_code(self, code_snippet):
        """代码审查"""
        issues = []
        
        # 检查常见问题模式
        patterns = {
            r'open\s*\([^)]+\)(?![^{]*with)': '文件未使用with语句',
            r'global\s+\w+.*=.*\[\]': '全局列表可能导致内存泄漏',
            r'\.append\(.*\)(?!.*\.pop|.*\.clear)': '只添加不清理的容器',
            r'threading\.Thread.*(?!.*\.join)': '线程未正确等待结束'
        }
        
        import re
        for pattern, message in patterns.items():
            if re.search(pattern, code_snippet):
                issues.append(message)
        
        return issues

# 使用示例
reviewer = MemoryLeakCodeReviewChecklist()

code_sample = """
def process_files(file_list):
    results = []
    for filename in file_list:
        f = open(filename)  # 问题：未使用with语句
        content = f.read()
        results.append(content)  # 问题：结果列表持续增长
    return results

global_cache = []  # 问题：全局列表

def cache_data(data):
    global_cache.append(data)  # 问题：只添加不清理
"""

issues = reviewer.review_code(code_sample)
for issue in issues:
    print(f"⚠️  {issue}")

import contextlib
import weakref
from abc import ABC, abstractmethod

class MemorySafeResource(ABC):
    """内存安全的资源基类"""
    
    def __init__(self):
        self._finalized = False
    
    @abstractmethod
    def _cleanup(self):
        """子类实现具体的清理逻辑"""
        pass
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()
    
    def close(self):
        """手动关闭资源"""
        if not self._finalized:
            self._cleanup()
            self._finalized = True
    
    def __del__(self):
        """确保资源在对象销毁时被清理"""
        self.close()

class SafeFileHandler(MemorySafeResource):
    """安全的文件处理器"""
    
    def __init__(self, filename, mode='r'):
        super().__init__()
        self.filename = filename
        self.file = open(filename, mode)
    
    def _cleanup(self):
        if self.file and not self.file.closed:
            self.file.close()
    
    def read(self):
        return self.file.read()
    
    def write(self, data):
        self.file.write(data)

class LRUCache:
    """内存安全的LRU缓存"""
    
    def __init__(self, max_size=1000):
        self.max_size = max_size
        self.cache = {}
        self.access_order = []
    
    def get(self, key):
        if key in self.cache:
            # 更新访问顺序
            self.access_order.remove(key)
            self.access_order.append(key)
            return self.cache[key]
        return None
    
    def put(self, key, value):
        if key in self.cache:
            # 更新现有键
            self.access_order.remove(key)
        elif len(self.cache) >= self.max_size:
            # 移除最久未使用的项
            oldest_key = self.access_order.pop(0)
            del self.cache[oldest_key]
        
        self.cache[key] = value
        self.access_order.append(key)
    
    def clear(self):
        """清空缓存"""
        self.cache.clear()
        self.access_order.clear()

class WeakCallbackRegistry:
    """弱引用回调注册器，避免循环引用"""
    
    def __init__(self):
        self._callbacks = weakref.WeakKeyDictionary()
    
    def register(self, obj, callback):
        """注册对象的回调函数"""
        if obj not in self._callbacks:
            self._callbacks[obj] = []
        self._callbacks[obj].append(callback)
    
    def notify(self, obj, *args, **kwargs):
        """通知对象的所有回调"""
        callbacks = self._callbacks.get(obj, [])
        for callback in callbacks:
            try:
                callback(*args, **kwargs)
            except Exception as e:
                print(f"回调执行失败: {e}")
    
    def unregister(self, obj):
        """注销对象的所有回调"""
        if obj in self._callbacks:
            del self._callbacks[obj]

# 使用示例
# 1. 安全的文件处理
with SafeFileHandler('test.txt', 'w') as handler:
    handler.write('Hello, World!')
# 文件自动关闭

# 2. 限制大小的缓存
cache = LRUCache(max_size=100)
for i in range(150):
    cache.put(f"key_{i}", f"value_{i}")
print(f"缓存大小: {len(cache.cache)}")  # 最多100个

# 3. 弱引用回调
registry = WeakCallbackRegistry()

class Publisher:
    def __init__(self):
        self.data = "some data"

class Subscriber:
    def callback(self, data):
        print(f"收到数据: {data}")

publisher = Publisher()
subscriber = Subscriber()
registry.register(publisher, subscriber.callback)
registry.notify(publisher, publisher.data)