Python内存优化的实战技巧分享

import sys

# 查看对象引用计数
a = [1, 2, 3]
print(f"引用计数: {sys.getrefcount(a)}")  # 输出: 2 (包括getrefcount的临时引用)

b = a  # 增加引用
print(f"引用计数: {sys.getrefcount(a)}")  # 输出: 3

del b  # 减少引用
print(f"引用计数: {sys.getrefcount(a)}")  # 输出: 2

import gc

# 查看垃圾回收统计信息
print(f"垃圾回收统计: {gc.get_stats()}")

# 手动触发垃圾回收
collected = gc.collect()
print(f"回收的对象数量: {collected}")

# 问题代码：循环引用导致内存泄漏
class node:
    def __init__(self, value):
        self.value = value
        self.parent = none
        self.children = []
    
    def add_child(self, child):
        child.parent = self  # 循环引用
        self.children.append(child)

# 优化方案：使用弱引用
import weakref

class optimizednode:
    def __init__(self, value):
        self.value = value
        self._parent = none
        self.children = []
    
    @property
    def parent(self):
        return self._parent() if self._parent else none
    
    @parent.setter
    def parent(self, value):
        self._parent = weakref.ref(value) if value else none
    
    def add_child(self, child):
        child.parent = self
        self.children.append(child)

# 问题代码：全局变量持续增长
global_cache = {}

def process_data(data):
    # 缓存持续增长，永不清理
    global_cache[data.id] = data
    return process(data)

# 优化方案：使用lru缓存
from functools import lru_cache
from collections import ordereddict

class lrucache:
    def __init__(self, max_size=1000):
        self.cache = ordereddict()
        self.max_size = max_size
    
    def get(self, key):
        if key in self.cache:
            # 移到末尾（最近使用）
            self.cache.move_to_end(key)
            return self.cache[key]
        return none
    
    def put(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        else:
            if len(self.cache) >= self.max_size:
                # 删除最久未使用的项
                self.cache.popitem(last=false)
        self.cache[key] = value

# 使用优化后的缓存
optimized_cache = lrucache(max_size=1000)

# 内存密集型：一次性加载所有数据
def read_large_file_bad(filename):
    with open(filename, 'r') as f:
        return f.readlines()  # 将整个文件加载到内存

# 内存优化：使用生成器
def read_large_file_good(filename):
    with open(filename, 'r') as f:
        for line in f:
            yield line.strip()

# 性能对比
import time
import psutil
import os

def measure_memory_usage(func, *args):
    process = psutil.process(os.getpid())
    start_memory = process.memory_info().rss / 1024 / 1024  # mb
    start_time = time.time()
    
    result = func(*args)
    
    end_time = time.time()
    end_memory = process.memory_info().rss / 1024 / 1024  # mb
    
    return {
        'result': result,
        'time': end_time - start_time,
        'memory_used': end_memory - start_memory
    }

# 普通类：使用字典存储属性
class regularpoint:
    def __init__(self, x, y):
        self.x = x
        self.y = y

# 优化类：使用__slots__
class optimizedpoint:
    __slots__ = ['x', 'y']
    
    def __init__(self, x, y):
        self.x = x
        self.y = y

# 内存使用对比
import sys

regular_point = regularpoint(1, 2)
optimized_point = optimizedpoint(1, 2)

print(f"普通类内存使用: {sys.getsizeof(regular_point.__dict__)} bytes")
print(f"优化类内存使用: {sys.getsizeof(optimized_point)} bytes")

# 批量创建对象的性能测试
def create_regular_points(n):
    return [regularpoint(i, i+1) for i in range(n)]

def create_optimized_points(n):
    return [optimizedpoint(i, i+1) for i in range(n)]

# 测试100万个对象的内存使用
n = 1000000
regular_stats = measure_memory_usage(create_regular_points, n)
optimized_stats = measure_memory_usage(create_optimized_points, n)

print(f"普通类 - 时间: {regular_stats['time']:.2f}s, 内存: {regular_stats['memory_used']:.2f}mb")
print(f"优化类 - 时间: {optimized_stats['time']:.2f}s, 内存: {optimized_stats['memory_used']:.2f}mb")

# 低效的字符串拼接
def inefficient_string_concat(items):
    result = ""
    for item in items:
        result += str(item) + ","
    return result[:-1]

# 高效的字符串拼接
def efficient_string_concat(items):
    return ",".join(str(item) for item in items)

# 使用字符串池优化
import sys

def string_interning_demo():
    # 小整数和短字符串会被自动intern
    a = "hello"
    b = "hello"
    print(f"字符串是否为同一对象: {a is b}")  # true
    
    # 手动intern长字符串
    long_str1 = sys.intern("this is a very long string that would not be interned automatically")
    long_str2 = sys.intern("this is a very long string that would not be interned automatically")
    print(f"长字符串是否为同一对象: {long_str1 is long_str2}")  # true

# 性能测试
items = list(range(10000))
inefficient_stats = measure_memory_usage(inefficient_string_concat, items)
efficient_stats = measure_memory_usage(efficient_string_concat, items)

print(f"低效拼接 - 时间: {inefficient_stats['time']:.4f}s")
print(f"高效拼接 - 时间: {efficient_stats['time']:.4f}s")
print(f"性能提升: {inefficient_stats['time'] / efficient_stats['time']:.2f}倍")

# 使用array替代list存储数值
import array

# 普通列表
regular_list = [i for i in range(1000000)]

# 数组（更节省内存）
int_array = array.array('i', range(1000000))

print(f"列表内存使用: {sys.getsizeof(regular_list)} bytes")
print(f"数组内存使用: {sys.getsizeof(int_array)} bytes")
print(f"内存节省: {(sys.getsizeof(regular_list) - sys.getsizeof(int_array)) / sys.getsizeof(regular_list) * 100:.1f}%")

# 使用collections.deque优化队列操作
from collections import deque

# 普通列表作为队列（低效）
def list_queue_operations(n):
    queue = []
    for i in range(n):
        queue.append(i)
    for i in range(n // 2):
        queue.pop(0)  # o(n)操作
    return queue

# deque作为队列（高效）
def deque_queue_operations(n):
    queue = deque()
    for i in range(n):
        queue.append(i)
    for i in range(n // 2):
        queue.popleft()  # o(1)操作
    return queue

# 性能对比
n = 50000
list_stats = measure_memory_usage(list_queue_operations, n)
deque_stats = measure_memory_usage(deque_queue_operations, n)

print(f"列表队列 - 时间: {list_stats['time']:.4f}s")
print(f"deque队列 - 时间: {deque_stats['time']:.4f}s")
print(f"性能提升: {list_stats['time'] / deque_stats['time']:.2f}倍")

import pandas as pd
import numpy as np
from typing import iterator

class dataprocessor:
    """大数据处理器 - 内存优化版本"""
    
    def __init__(self, chunk_size: int = 10000):
        self.chunk_size = chunk_size
    
    def process_large_csv(self, filename: str) -> iterator[pd.dataframe]:
        """分块处理大型csv文件"""
        for chunk in pd.read_csv(filename, chunksize=self.chunk_size):
            # 优化数据类型
            chunk = self._optimize_dtypes(chunk)
            yield self._process_chunk(chunk)
    
    def _optimize_dtypes(self, df: pd.dataframe) -> pd.dataframe:
        """优化dataframe的数据类型以节省内存"""
        for col in df.columns:
            col_type = df[col].dtype
            
            if col_type != 'object':
                c_min = df[col].min()
                c_max = df[col].max()
                
                if str(col_type)[:3] == 'int':
                    if c_min > np.iinfo(np.int8).min and c_max < np.iinfo(np.int8).max:
                        df[col] = df[col].astype(np.int8)
                    elif c_min > np.iinfo(np.int16).min and c_max < np.iinfo(np.int16).max:
                        df[col] = df[col].astype(np.int16)
                    elif c_min > np.iinfo(np.int32).min and c_max < np.iinfo(np.int32).max:
                        df[col] = df[col].astype(np.int32)
                
                elif str(col_type)[:5] == 'float':
                    if c_min > np.finfo(np.float32).min and c_max < np.finfo(np.float32).max:
                        df[col] = df[col].astype(np.float32)
        
        return df
    
    def _process_chunk(self, chunk: pd.dataframe) -> pd.dataframe:
        """处理数据块"""
        # 示例处理逻辑
        chunk['processed'] = chunk.sum(axis=1, numeric_only=true)
        return chunk
    
    def get_memory_usage(self, df: pd.dataframe) -> dict:
        """获取dataframe内存使用情况"""
        return {
            'total_memory': df.memory_usage(deep=true).sum(),
            'memory_per_column': df.memory_usage(deep=true).to_dict()
        }

# 使用示例
processor = dataprocessor(chunk_size=5000)

# 模拟处理大文件
def simulate_large_data_processing():
    # 创建测试数据
    test_data = pd.dataframe({
        'id': range(100000),
        'value1': np.random.randint(0, 1000, 100000),
        'value2': np.random.random(100000),
        'category': np.random.choice(['a', 'b', 'c'], 100000)
    })
    
    # 保存为csv
    test_data.to_csv('test_large_data.csv', index=false)
    
    # 处理数据
    results = []
    for processed_chunk in processor.process_large_csv('test_large_data.csv'):
        results.append(processed_chunk)
    
    return pd.concat(results, ignore_index=true)

# 性能测试
large_data_stats = measure_memory_usage(simulate_large_data_processing)
print(f"大数据处理 - 时间: {large_data_stats['time']:.2f}s, 内存: {large_data_stats['memory_used']:.2f}mb")

import threading
import time
from typing import any, optional
from dataclasses import dataclass

@dataclass
class cacheitem:
    """缓存项"""
    value: any
    timestamp: float
    access_count: int = 0
    
    def is_expired(self, ttl: float) -> bool:
        return time.time() - self.timestamp > ttl

class memoryefficientcache:
    """内存高效的缓存系统"""
    
    def __init__(self, max_size: int = 1000, ttl: float = 3600):
        self.max_size = max_size
        self.ttl = ttl
        self._cache = {}
        self._lock = threading.rlock()
        self._access_order = []
    
    def get(self, key: str) -> optional[any]:
        with self._lock:
            if key not in self._cache:
                return none
            
            item = self._cache[key]
            
            # 检查是否过期
            if item.is_expired(self.ttl):
                del self._cache[key]
                if key in self._access_order:
                    self._access_order.remove(key)
                return none
            
            # 更新访问信息
            item.access_count += 1
            if key in self._access_order:
                self._access_order.remove(key)
            self._access_order.append(key)
            
            return item.value
    
    def put(self, key: str, value: any) -> none:
        with self._lock:
            # 如果缓存已满，移除最少使用的项
            if len(self._cache) >= self.max_size and key not in self._cache:
                self._evict_lru()
            
            # 添加或更新缓存项
            self._cache[key] = cacheitem(value, time.time())
            if key in self._access_order:
                self._access_order.remove(key)
            self._access_order.append(key)
    
    def _evict_lru(self) -> none:
        """移除最少使用的缓存项"""
        if not self._access_order:
            return
        
        lru_key = self._access_order.pop(0)
        if lru_key in self._cache:
            del self._cache[lru_key]
    
    def clear_expired(self) -> int:
        """清理过期的缓存项"""
        with self._lock:
            expired_keys = [
                key for key, item in self._cache.items()
                if item.is_expired(self.ttl)
            ]
            
            for key in expired_keys:
                del self._cache[key]
                if key in self._access_order:
                    self._access_order.remove(key)
            
            return len(expired_keys)
    
    def get_stats(self) -> dict:
        """获取缓存统计信息"""
        with self._lock:
            return {
                'size': len(self._cache),
                'max_size': self.max_size,
                'hit_rate': self._calculate_hit_rate(),
                'memory_usage': sum(sys.getsizeof(item.value) for item in self._cache.values())
            }
    
    def _calculate_hit_rate(self) -> float:
        """计算缓存命中率"""
        total_access = sum(item.access_count for item in self._cache.values())
        return total_access / len(self._cache) if self._cache else 0.0

# 缓存性能测试
def test_cache_performance():
    cache = memoryefficientcache(max_size=1000, ttl=60)
    
    # 写入测试
    start_time = time.time()
    for i in range(5000):
        cache.put(f"key_{i}", f"value_{i}" * 100)  # 较大的值
    write_time = time.time() - start_time
    
    # 读取测试
    start_time = time.time()
    hits = 0
    for i in range(5000):
        if cache.get(f"key_{i % 1000}") is not none:  # 部分命中
            hits += 1
    read_time = time.time() - start_time
    
    stats = cache.get_stats()
    
    return {
        'write_time': write_time,
        'read_time': read_time,
        'hit_rate': hits / 5000,
        'cache_stats': stats
    }

cache_performance = test_cache_performance()
print(f"缓存写入时间: {cache_performance['write_time']:.4f}s")
print(f"缓存读取时间: {cache_performance['read_time']:.4f}s")
print(f"缓存命中率: {cache_performance['hit_rate']:.2%}")
print(f"缓存内存使用: {cache_performance['cache_stats']['memory_usage'] / 1024 / 1024:.2f}mb")

import tracemalloc
import linecache
import gc
from typing import list, tuple

class memoryprofiler:
    """内存分析器"""
    
    def __init__(self):
        self.snapshots = []
    
    def start_tracing(self):
        """开始内存追踪"""
        tracemalloc.start()
    
    def take_snapshot(self, description: str = ""):
        """拍摄内存快照"""
        snapshot = tracemalloc.take_snapshot()
        self.snapshots.append((description, snapshot))
        return snapshot
    
    def compare_snapshots(self, snapshot1_idx: int = 0, snapshot2_idx: int = -1) -> list[tuple]:
        """比较两个快照"""
        if len(self.snapshots) < 2:
            return []
        
        _, snapshot1 = self.snapshots[snapshot1_idx]
        _, snapshot2 = self.snapshots[snapshot2_idx]
        
        top_stats = snapshot2.compare_to(snapshot1, 'lineno')
        return top_stats[:10]  # 返回前10个差异最大的
    
    def get_top_memory_usage(self, snapshot_idx: int = -1, limit: int = 10) -> list:
        """获取内存使用最多的代码行"""
        if not self.snapshots:
            return []
        
        _, snapshot = self.snapshots[snapshot_idx]
        top_stats = snapshot.statistics('lineno')
        
        result = []
        for stat in top_stats[:limit]:
            frame = stat.traceback.format()[-1]
            result.append({
                'memory': stat.size,
                'memory_mb': stat.size / 1024 / 1024,
                'count': stat.count,
                'frame': frame
            })
        
        return result
    
    def analyze_memory_leaks(self) -> dict:
        """分析内存泄漏"""
        if len(self.snapshots) < 2:
            return {}
        
        # 比较第一个和最后一个快照
        top_stats = self.compare_snapshots(0, -1)
        
        potential_leaks = []
        for stat in top_stats:
            if stat.size_diff > 1024 * 1024:  # 增长超过1mb
                potential_leaks.append({
                    'size_diff_mb': stat.size_diff / 1024 / 1024,
                    'count_diff': stat.count_diff,
                    'traceback': stat.traceback.format()
                })
        
        return {
            'total_snapshots': len(self.snapshots),
            'potential_leaks': potential_leaks,
            'gc_stats': gc.get_stats()
        }

# 使用示例
def memory_intensive_function():
    """内存密集型函数示例"""
    data = []
    for i in range(100000):
        data.append([j for j in range(100)])
    return data

def optimized_memory_function():
    """优化后的内存函数"""
    for i in range(100000):
        yield [j for j in range(100)]

# 内存分析
profiler = memoryprofiler()
profiler.start_tracing()

# 第一个快照
profiler.take_snapshot("开始")

# 执行内存密集型操作
data1 = memory_intensive_function()
profiler.take_snapshot("内存密集型函数执行后")

# 清理数据
del data1
gc.collect()
profiler.take_snapshot("清理后")

# 执行优化后的操作
data2 = list(optimized_memory_function())
profiler.take_snapshot("优化函数执行后")

# 分析结果
leak_analysis = profiler.analyze_memory_leaks()
top_usage = profiler.get_top_memory_usage()

print("=== 内存使用分析 ===")
for usage in top_usage[:5]:
    print(f"内存: {usage['memory_mb']:.2f}mb, 调用次数: {usage['count']}")
    print(f"位置: {usage['frame']}")
    print("-" * 50)

print("\n=== 潜在内存泄漏 ===")
for leak in leak_analysis.get('potential_leaks', []):
    print(f"内存增长: {leak['size_diff_mb']:.2f}mb")
    print(f"对象增长: {leak['count_diff']}")
    print("-" * 50)

import psutil
import threading
import time
from collections import deque
from typing import dict, list

class realtimemonitor:
    """实时内存监控器"""
    
    def __init__(self, interval: float = 1.0, history_size: int = 100):
        self.interval = interval
        self.history_size = history_size
        self.monitoring = false
        self.monitor_thread = none
        
        # 历史数据
        self.memory_history = deque(maxlen=history_size)
        self.cpu_history = deque(maxlen=history_size)
        self.timestamp_history = deque(maxlen=history_size)
    
    def start_monitoring(self):
        """开始监控"""
        if self.monitoring:
            return
        
        self.monitoring = true
        self.monitor_thread = threading.thread(target=self._monitor_loop, daemon=true)
        self.monitor_thread.start()
    
    def stop_monitoring(self):
        """停止监控"""
        self.monitoring = false
        if self.monitor_thread:
            self.monitor_thread.join()
    
    def _monitor_loop(self):
        """监控循环"""
        process = psutil.process()
        
        while self.monitoring:
            try:
                # 获取当前系统信息
                memory_info = process.memory_info()
                cpu_percent = process.cpu_percent()
                
                # 记录数据
                self.memory_history.append(memory_info.rss / 1024 / 1024)  # mb
                self.cpu_history.append(cpu_percent)
                self.timestamp_history.append(time.time())
                
                time.sleep(self.interval)
                
            except exception as e:
                print(f"监控错误: {e}")
                break
    
    def get_current_stats(self) -> dict:
        """获取当前统计信息"""
        if not self.memory_history:
            return {}
        
        return {
            'current_memory_mb': self.memory_history[-1],
            'current_cpu_percent': self.cpu_history[-1],
            'avg_memory_mb': sum(self.memory_history) / len(self.memory_history),
            'max_memory_mb': max(self.memory_history),
            'min_memory_mb': min(self.memory_history),
            'memory_trend': self._calculate_trend(self.memory_history)
        }
    
    def _calculate_trend(self, data: deque) -> str:
        """计算趋势"""
        if len(data) < 10:
            return "insufficient_data"
        
        recent = list(data)[-10:]
        earlier = list(data)[-20:-10] if len(data) >= 20 else list(data)[:-10]
        
        if not earlier:
            return "insufficient_data"
        
        recent_avg = sum(recent) / len(recent)
        earlier_avg = sum(earlier) / len(earlier)
        
        diff_percent = (recent_avg - earlier_avg) / earlier_avg * 100
        
        if diff_percent > 5:
            return "increasing"
        elif diff_percent < -5:
            return "decreasing"
        else:
            return "stable"
    
    def export_data(self) -> dict[str, list]:
        """导出监控数据"""
        return {
            'timestamps': list(self.timestamp_history),
            'memory_mb': list(self.memory_history),
            'cpu_percent': list(self.cpu_history)
        }

# 监控使用示例
def test_with_monitoring():
    monitor = realtimemonitor(interval=0.5)
    monitor.start_monitoring()
    
    try:
        # 模拟一些内存操作
        print("开始内存密集型操作...")
        
        # 创建大量对象
        large_list = []
        for i in range(50000):
            large_list.append([j for j in range(100)])
            
            if i % 10000 == 0:
                stats = monitor.get_current_stats()
                print(f"进度: {i/50000*100:.1f}%, "
                      f"内存: {stats.get('current_memory_mb', 0):.1f}mb, "
                      f"趋势: {stats.get('memory_trend', 'unknown')}")
                time.sleep(0.1)
        
        print("操作完成，等待5秒...")
        time.sleep(5)
        
        # 清理内存
        del large_list
        gc.collect()
        
        print("内存清理完成，等待5秒...")
        time.sleep(5)
        
    finally:
        monitor.stop_monitoring()
        
        # 输出最终统计
        final_stats = monitor.get_current_stats()
        print("\n=== 最终统计 ===")
        for key, value in final_stats.items():
            print(f"{key}: {value}")

# 运行监控测试
test_with_monitoring()

# ✅ 推荐做法
class optimizedclass:
    __slots__ = ['x', 'y', 'z']  # 使用__slots__
    
    def __init__(self, x, y, z):
        self.x = x
        self.y = y
        self.z = z
    
    def process_data(self, data):
        # 使用生成器表达式
        return (item * 2 for item in data if item > 0)
    
    def string_operations(self, items):
        # 使用join而不是+=
        return ''.join(str(item) for item in items)

# ❌ 避免的做法
class regularclass:
    def __init__(self, x, y, z):
        self.x = x
        self.y = y
        self.z = z
        self.cache = {}  # 可能导致内存泄漏
    
    def process_data(self, data):
        # 创建完整列表
        return [item * 2 for item in data if item > 0]
    
    def string_operations(self, items):
        # 低效的字符串拼接
        result = ""
        for item in items:
            result += str(item)
        return result

from collections import deque, defaultdict, counter
import array

# 根据使用场景选择合适的数据结构
def choose_right_data_structure():
    # 队列操作：使用deque
    queue = deque()
    
    # 数值数组：使用array
    numbers = array.array('i', range(1000))
    
    # 计数操作：使用counter
    counter = counter(['a', 'b', 'a', 'c', 'b', 'a'])
    
    # 默认值字典：使用defaultdict
    grouped_data = defaultdict(list)
    
    return queue, numbers, counter, grouped_data

def memory_optimization_checklist():
    """内存优化检查清单"""
    checklist = {
        "代码优化": [
            "✓ 使用生成器替代大列表",
            "✓ 为频繁创建的类添加__slots__",
            "✓ 使用join()进行字符串拼接",
            "✓ 及时删除不需要的大对象",
            "✓ 避免循环引用"
        ],
        "数据结构": [
            "✓ 选择合适的数据类型（array vs list）",
            "✓ 使用deque进行队列操作",
            "✓ 考虑使用numpy处理数值计算",
            "✓ 实现lru缓存避免无限增长"
        ],
        "监控工具": [
            "✓ 使用tracemalloc追踪内存分配",
            "✓ 定期检查gc.get_stats()",
            "✓ 监控进程内存使用情况",
            "✓ 分析内存增长趋势"
        ],
        "最佳实践": [
            "✓ 分块处理大文件",
            "✓ 使用上下文管理器确保资源释放",
            "✓ 定期清理过期缓存",
            "✓ 在生产环境中持续监控"
        ]
    }
    
    for category, items in checklist.items():
        print(f"\n{category}:")
        for item in items:
            print(f"  {item}")

memory_optimization_checklist()