import sys

# 查看對象引用計數(shù)
a = [1, 2, 3]
print(f"引用計數(shù): {sys.getrefcount(a)}")  # 輸出: 2 (包括getrefcount的臨時引用)

b = a  # 增加引用
print(f"引用計數(shù): {sys.getrefcount(a)}")  # 輸出: 3

del b  # 減少引用
print(f"引用計數(shù): {sys.getrefcount(a)}")  # 輸出: 2

import gc

# 查看垃圾回收統(tǒng)計信息
print(f"垃圾回收統(tǒng)計: {gc.get_stats()}")

# 手動觸發(fā)垃圾回收
collected = gc.collect()
print(f"回收的對象數(shù)量: {collected}")

# 問題代碼：循環(huán)引用導致內(nèi)存泄漏
class Node:
    def __init__(self, value):
        self.value = value
        self.parent = None
        self.children = []
    
    def add_child(self, child):
        child.parent = self  # 循環(huán)引用
        self.children.append(child)

# 優(yōu)化方案：使用弱引用
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)

# 問題代碼：全局變量持續(xù)增長
global_cache = {}

def process_data(data):
    # 緩存持續(xù)增長，永不清理
    global_cache[data.id] = data
    return process(data)

# 優(yōu)化方案：使用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

# 使用優(yōu)化后的緩存
optimized_cache = LRUCache(max_size=1000)

# 內(nèi)存密集型：一次性加載所有數(shù)據(jù)
def read_large_file_bad(filename):
    with open(filename, 'r') as f:
        return f.readlines()  # 將整個文件加載到內(nèi)存

# 內(nèi)存優(yōu)化：使用生成器
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

# 優(yōu)化類：使用__slots__
class OptimizedPoint:
    __slots__ = ['x', 'y']
    
    def __init__(self, x, y):
        self.x = x
        self.y = y

# 內(nèi)存使用對比
import sys

regular_point = RegularPoint(1, 2)
optimized_point = OptimizedPoint(1, 2)

print(f"普通類內(nèi)存使用: {sys.getsizeof(regular_point.__dict__)} bytes")
print(f"優(yōu)化類內(nèi)存使用: {sys.getsizeof(optimized_point)} bytes")

# 批量創(chuàng)建對象的性能測試
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èi)存使用
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, 內(nèi)存: {regular_stats['memory_used']:.2f}MB")
print(f"優(yōu)化類 - 時間: {optimized_stats['time']:.2f}s, 內(nèi)存: {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)

# 使用字符串池優(yōu)化
import sys

def string_interning_demo():
    # 小整數(shù)和短字符串會被自動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存儲數(shù)值
import array

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

# 數(shù)組（更節(jié)省內(nèi)存）
int_array = array.array('i', range(1000000))

print(f"列表內(nèi)存使用: {sys.getsizeof(regular_list)} bytes")
print(f"數(shù)組內(nèi)存使用: {sys.getsizeof(int_array)} bytes")
print(f"內(nèi)存節(jié)省: {(sys.getsizeof(regular_list) - sys.getsizeof(int_array)) / sys.getsizeof(regular_list) * 100:.1f}%")

# 使用collections.deque優(yōu)化隊列操作
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:
    """大數(shù)據(jù)處理器 - 內(nèi)存優(yōu)化版本"""
    
    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):
            # 優(yōu)化數(shù)據(jù)類型
            chunk = self._optimize_dtypes(chunk)
            yield self._process_chunk(chunk)
    
    def _optimize_dtypes(self, df: pd.DataFrame) -> pd.DataFrame:
        """優(yōu)化DataFrame的數(shù)據(jù)類型以節(jié)省內(nèi)存"""
        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:
        """處理數(shù)據(jù)塊"""
        # 示例處理邏輯
        chunk['processed'] = chunk.sum(axis=1, numeric_only=True)
        return chunk
    
    def get_memory_usage(self, df: pd.DataFrame) -> dict:
        """獲取DataFrame內(nèi)存使用情況"""
        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():
    # 創(chuàng)建測試數(shù)據(jù)
    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)
    
    # 處理數(shù)據(jù)
    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"大數(shù)據(jù)處理 - 時間: {large_data_stats['time']:.2f}s, 內(nèi)存: {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:
    """內(nèi)存高效的緩存系統(tǒng)"""
    
    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:
        """獲取緩存統(tǒng)計信息"""
        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"緩存內(nèi)存使用: {cache_performance['cache_stats']['memory_usage'] / 1024 / 1024:.2f}MB")

import tracemalloc
import linecache
import gc
from typing import List, Tuple

class MemoryProfiler:
    """內(nèi)存分析器"""
    
    def __init__(self):
        self.snapshots = []
    
    def start_tracing(self):
        """開始內(nèi)存追蹤"""
        tracemalloc.start()
    
    def take_snapshot(self, description: str = ""):
        """拍攝內(nèi)存快照"""
        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:
        """獲取內(nèi)存使用最多的代碼行"""
        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:
        """分析內(nèi)存泄漏"""
        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():
    """內(nèi)存密集型函數(shù)示例"""
    data = []
    for i in range(100000):
        data.append([j for j in range(100)])
    return data

def optimized_memory_function():
    """優(yōu)化后的內(nèi)存函數(shù)"""
    for i in range(100000):
        yield [j for j in range(100)]

# 內(nèi)存分析
profiler = MemoryProfiler()
profiler.start_tracing()

# 第一個快照
profiler.take_snapshot("開始")

# 執(zhí)行內(nèi)存密集型操作
data1 = memory_intensive_function()
profiler.take_snapshot("內(nèi)存密集型函數(shù)執(zhí)行后")

# 清理數(shù)據(jù)
del data1
gc.collect()
profiler.take_snapshot("清理后")

# 執(zhí)行優(yōu)化后的操作
data2 = list(optimized_memory_function())
profiler.take_snapshot("優(yōu)化函數(shù)執(zhí)行后")

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

print("=== 內(nèi)存使用分析 ===")
for usage in top_usage[:5]:
    print(f"內(nèi)存: {usage['memory_mb']:.2f}MB, 調(diào)用次數(shù): {usage['count']}")
    print(f"位置: {usage['frame']}")
    print("-" * 50)

print("\n=== 潛在內(nèi)存泄漏 ===")
for leak in leak_analysis.get('potential_leaks', []):
    print(f"內(nèi)存增長: {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:
    """實時內(nèi)存監(jiān)控器"""
    
    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
        
        # 歷史數(shù)據(jù)
        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):
        """開始監(jiān)控"""
        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):
        """停止監(jiān)控"""
        self.monitoring = False
        if self.monitor_thread:
            self.monitor_thread.join()
    
    def _monitor_loop(self):
        """監(jiān)控循環(huán)"""
        process = psutil.Process()
        
        while self.monitoring:
            try:
                # 獲取當前系統(tǒng)信息
                memory_info = process.memory_info()
                cpu_percent = process.cpu_percent()
                
                # 記錄數(shù)據(jù)
                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"監(jiān)控錯誤: {e}")
                break
    
    def get_current_stats(self) -> Dict:
        """獲取當前統(tǒng)計信息"""
        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]:
        """導出監(jiān)控數(shù)據(jù)"""
        return {
            'timestamps': list(self.timestamp_history),
            'memory_mb': list(self.memory_history),
            'cpu_percent': list(self.cpu_history)
        }

# 監(jiān)控使用示例
def test_with_monitoring():
    monitor = RealTimeMonitor(interval=0.5)
    monitor.start_monitoring()
    
    try:
        # 模擬一些內(nèi)存操作
        print("開始內(nèi)存密集型操作...")
        
        # 創(chuàng)建大量對象
        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"內(nèi)存: {stats.get('current_memory_mb', 0):.1f}MB, "
                      f"趨勢: {stats.get('memory_trend', 'unknown')}")
                time.sleep(0.1)
        
        print("操作完成，等待5秒...")
        time.sleep(5)
        
        # 清理內(nèi)存
        del large_list
        gc.collect()
        
        print("內(nèi)存清理完成，等待5秒...")
        time.sleep(5)
        
    finally:
        monitor.stop_monitoring()
        
        # 輸出最終統(tǒng)計
        final_stats = monitor.get_current_stats()
        print("\n=== 最終統(tǒng)計 ===")
        for key, value in final_stats.items():
            print(f"{key}: {value}")

# 運行監(jiān)控測試
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 = {}  # 可能導致內(nèi)存泄漏
    
    def process_data(self, data):
        # 創(chuàng)建完整列表
        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

# 根據(jù)使用場景選擇合適的數(shù)據(jù)結構
def choose_right_data_structure():
    # 隊列操作：使用deque
    queue = deque()
    
    # 數(shù)值數(shù)組：使用array
    numbers = array.array('i', range(1000))
    
    # 計數(shù)操作：使用Counter
    counter = Counter(['a', 'b', 'a', 'c', 'b', 'a'])
    
    # 默認值字典：使用defaultdict
    grouped_data = defaultdict(list)
    
    return queue, numbers, counter, grouped_data

def memory_optimization_checklist():
    """內(nèi)存優(yōu)化檢查清單"""
    checklist = {
        "代碼優(yōu)化": [
            "? 使用生成器替代大列表",
            "? 為頻繁創(chuàng)建的類添加__slots__",
            "? 使用join()進行字符串拼接",
            "? 及時刪除不需要的大對象",
            "? 避免循環(huán)引用"
        ],
        "數(shù)據(jù)結構": [
            "? 選擇合適的數(shù)據(jù)類型（array vs list）",
            "? 使用deque進行隊列操作",
            "? 考慮使用numpy處理數(shù)值計算",
            "? 實現(xiàn)LRU緩存避免無限增長"
        ],
        "監(jiān)控工具": [
            "? 使用tracemalloc追蹤內(nèi)存分配",
            "? 定期檢查gc.get_stats()",
            "? 監(jiān)控進程內(nèi)存使用情況",
            "? 分析內(nèi)存增長趨勢"
        ],
        "最佳實踐": [
            "? 分塊處理大文件",
            "? 使用上下文管理器確保資源釋放",
            "? 定期清理過期緩存",
            "? 在生產(chǎn)環(huán)境中持續(xù)監(jiān)控"
        ]
    }
    
    for category, items in checklist.items():
        print(f"\n{category}:")
        for item in items:
            print(f"  {item}")

memory_optimization_checklist()