5.3 内存分析工具

python

# 安装内存分析工具

# pip install memory_profiler

# pip install objgraph

try:

   import memory_profiler

   import objgraph

   import random

   

   def memory_intensive_function():

       """内存密集型函数"""

       # 创建大量对象

       big_list = []

       for i in range(10000):

           # 创建复杂对象

           obj = {

               'id': i,

               'data': [random.random() for _ in range(100)],

               'nested': {

                   'x': random.randint(1, 100),

                   'y': random.randint(1, 100),

                   'z': random.randint(1, 100)

               }

           }

           big_list.append(obj)

       

       # 处理数据

       result = []

       for obj in big_list:

           processed = {

               'id': obj['id'],

               'sum': sum(obj['data']),

               'product': obj['nested']['x'] * obj['nested']['y'] * obj['nested']['z']

           }

           result.append(processed)

       

       return result

   

   def analyze_memory():

       """分析内存使用"""

       print("内存分析示例:")

       print("=" * 50)

       

       # 使用memory_profiler

       print("内存使用分析:")

       

       # 获取基线内存

       mem_before = memory_profiler.memory_usage()[0]

       print(f"执行前内存: {mem_before:.2f} MB")

       

       # 执行函数

       result = memory_intensive_function()

       

       # 获取执行后内存

       mem_after = memory_profiler.memory_usage()[0]

       print(f"执行后内存: {mem_after:.2f} MB")

       print(f"内存增加: {mem_after - mem_before:.2f} MB")

       

       # 使用objgraph分析对象

       print("\n对象类型统计:")

       objgraph.show_most_common_types(limit=10)

       

       # 查找循环引用

       print("\n查找循环引用...")

       cycles = objgraph.find_backref_chain(

           random.choice(objgraph.by_type('dict')),

           objgraph.is_proper_module

       )

       

       if cycles:

           print(f"发现循环引用，长度: {len(cycles)}")

       else:

           print("未发现循环引用")

       

       # 清理

       del result

       import gc

       gc.collect()

   

   # 运行内存分析

   analyze_memory()

   

except ImportError as e:

   print(f"需要安装内存分析工具: {e}")

   print("运行: pip install memory_profiler objgraph")

第六部分：设计模式深入

6.1 观察者模式（Observer Pattern）

python

from abc import ABC, abstractmethod

from typing import List

# 观察者接口

class Observer(ABC):

   @abstractmethod

   def update(self, subject):

       pass

# 主题（被观察者）

class Subject:

   def __init__(self):

       self._observers: List[Observer] = []

       self._state = None

   

   def attach(self, observer: Observer):

       if observer not in self._observers:

           self._observers.append(observer)

           print(f"添加观察者: {observer}")

   

   def detach(self, observer: Observer):

       if observer in self._observers:

           self._observers.remove(observer)

           print(f"移除观察者: {observer}")

   

   def notify(self):

       print(f"通知 {len(self._observers)} 个观察者...")

       for observer in self._observers:

           observer.update(self)

   

   @property

   def state(self):

       return self._state

   

   @state.setter

   def state(self, value):

       self._state = value

       self.notify()

# 具体观察者

class EmailNotification(Observer):

   def __init__(self, name):

       self.name = name

   

   def update(self, subject):

       print(f"[{self.name}] 收到状态更新: {subject.state}")

       # 模拟发送邮件

       print(f"    发送邮件: 状态已更新为 {subject.state}")

   

   def __str__(self):

       return f"EmailNotification({self.name})"

class SMSNotification(Observer):

   def __init__(self, name):

       self.name = name

   

   def update(self, subject):

       print(f"[{self.name}] 收到状态更新: {subject.state}")

       # 模拟发送短信

       print(f"    发送短信: 状态: {subject.state}")

   

   def __str__(self):

       return f"SMSNotification({self.name})"

class LogNotification(Observer):

   def __init__(self, name):

       self.name = name

   

   def update(self, subject):

       print(f"[{self.name}] 收到状态更新: {subject.state}")

       # 模拟记录日志

       print(f"    记录日志: {subject.__class__.__name__} 状态变为 {subject.state}")

   

   def __str__(self):

       return f"LogNotification({self.name})"

# 使用示例

print("观察者模式示例:")

print("=" * 50)

# 创建主题

weather_station = Subject()

# 创建观察者

email_observer = EmailNotification("天气邮件通知")

sms_observer = SMSNotification("天气短信通知")

log_observer = LogNotification("系统日志")

# 注册观察者

weather_station.attach(email_observer)

weather_station.attach(sms_observer)

weather_station.attach(log_observer)

# 更新状态（自动通知所有观察者）

print("\n更新天气状态:")

weather_station.state = "晴天"

print("\n移除一个观察者后更新:")

weather_station.detach(sms_observer)

weather_station.state = "多云"

6.2 策略模式（Strategy Pattern）

python

from abc import ABC, abstractmethod

from typing import List

# 策略接口

class PaymentStrategy(ABC):

   @abstractmethod

   def pay(self, amount: float) -> str:

       pass

# 具体策略

class CreditCardPayment(PaymentStrategy):

   def __init__(self, card_number: str, cvv: str):

       self.card_number = card_number

       self.cvv = cvv

   

   def pay(self, amount: float) -> str:

       # 模拟信用卡支付

       return f"信用卡支付 {amount:.2f} 元 (卡号: {self.card_number[-4:]})"

class PayPalPayment(PaymentStrategy):

   def __init__(self, email: str):

       self.email = email

   

   def pay(self, amount: float) -> str:

       # 模拟PayPal支付

       return f"PayPal支付 {amount:.2f} 元 (邮箱: {self.email})"

class BitcoinPayment(PaymentStrategy):

   def __init__(self, wallet_address: str):

       self.wallet_address = wallet_address

   

   def pay(self, amount: float) -> str:

       # 模拟比特币支付

       return f"比特币支付 {amount:.2f} 元 (钱包地址: {self.wallet_address[:10]}...)"

# 上下文

class ShoppingCart:

   def __init__(self):

       self.items: List[dict] = []

       self.payment_strategy: PaymentStrategy = None

   

   def add_item(self, name: str, price: float):

       self.items.append({'name': name, 'price': price})

       print(f"添加商品: {name}, 价格: {price:.2f}")

   

   def calculate_total(self) -> float:

       return sum(item['price'] for item in self.items)

   

   def set_payment_strategy(self, strategy: PaymentStrategy):

       self.payment_strategy = strategy

       print(f"设置支付方式: {strategy.__class__.__name__}")

   

   def checkout(self) -> str:

       if not self.payment_strategy:

           return "错误: 未设置支付方式"

       

       total = self.calculate_total()

       if total <= 0:

           return "错误: 购物车为空"

       

       print(f"\n开始结账，总金额: {total:.2f}")

       result = self.payment_strategy.pay(total)

       

       # 清空购物车

       self.items.clear()

       

       return f"支付成功！{result}"

# 使用示例

print("\n策略模式示例:")

print("=" * 50)

# 创建购物车

cart = ShoppingCart()

# 添加商品

cart.add_item("Python编程书", 89.99)

cart.add_item("USB数据线", 19.99)

cart.add_item("无线鼠标", 129.99)

# 选择不同的支付策略

print("\n使用信用卡支付:")

cart.set_payment_strategy(

   CreditCardPayment("1234-5678-9012-3456", "123")

)

print(cart.checkout())

# 重新添加商品

print("\n\n新的购物:")

cart.add_item("笔记本电脑", 5999.99)

cart.add_item("电脑包", 199.99)

print("\n使用PayPal支付:")

cart.set_payment_strategy(

   PayPalPayment("user@example.com")

)

print(cart.checkout())

print("\n\n更多商品:")

cart.add_item("手机", 2999.99)

print("\n使用比特币支付:")

cart.set_payment_strategy(

   BitcoinPayment("1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa")

)

print(cart.checkout())

6.3 工厂模式（Factory Pattern）

python

from abc import ABC, abstractmethod

from enum import Enum

import json

import xml.etree.ElementTree as ET

# 产品接口

class DataFormatter(ABC):

   @abstractmethod

   def format(self, data: dict) -> str:

       pass

# 具体产品

class JSONFormatter(DataFormatter):

   def format(self, data: dict) -> str:

       return json.dumps(data, indent=2)

class XMLFormatter(DataFormatter):

   def format(self, data: dict) -> str:

       root = ET.Element("data")

       

       for key, value in data.items():

           element = ET.SubElement(root, key)

           element.text = str(value)

       

       # 格式化XML

       import xml.dom.minidom

       xml_str = ET.tostring(root, encoding='unicode')

       dom = xml.dom.minidom.parseString(xml_str)

       return dom.toprettyxml(indent="  ")

class CSVFormatter(DataFormatter):

   def format(self, data: dict) -> str:

       lines = []

       lines.append(",".join(data.keys()))

       lines.append(",".join(str(v) for v in data.values()))

       return "\n".join(lines)

class HTMLFormatter(DataFormatter):

   def format(self, data: dict) -> str:

       html = ["<table border='1'>"]

       html.append("  <tr>")

       

       for key in data.keys():

           html.append(f"    <th>{key}</th>")

       

       html.append("  </tr>")

       html.append("  <tr>")

       

       for value in data.values():

           html.append(f"    <td>{value}</td>")

       

       html.append("  </tr>")

       html.append("</table>")

       return "\n".join(html)

# 工厂

class FormatterFactory:

   @staticmethod

   def create_formatter(format_type: str) -> DataFormatter:

       formatters = {

           'json': JSONFormatter,

           'xml': XMLFormatter,

           'csv': CSVFormatter,

           'html': HTMLFormatter

       }

       

       if format_type not in formatters:

           raise ValueError(f"不支持格式: {format_type}")

       

       return formatters[format_type]()

# 客户端代码

class DataExporter:

   def __init__(self, data: dict):

       self.data = data

       self.formatter = None

   

   def set_format(self, format_type: str):

       self.formatter = FormatterFactory.create_formatter(format_type)

       print(f"设置为 {format_type.upper()} 格式")

   

   def export(self) -> str:

       if not self.formatter:

           raise ValueError("未设置格式")

       

       return self.formatter.format(self.data)

# 使用示例

print("\n工厂模式示例:")

print("=" * 50)

# 示例数据

sample_data = {

   "name": "张三",

   "age": 30,

   "email": "zhangsan@example.com",

   "city": "北京",

   "salary": 15000.50

}

# 创建导出器

exporter = DataExporter(sample_data)

# 测试不同格式

formats = ['json', 'xml', 'csv', 'html']

for fmt in formats:

   print(f"\n导出为 {fmt.upper()} 格式:")

   print("-" * 30)

   

   try:

       exporter.set_format(fmt)

       result = exporter.export()

       print(result[:200] + "..." if len(result) > 200 else result)

   except ValueError as e:

       print(f"错误: {e}")

# 动态添加新格式

print("\n\n扩展工厂模式:")

print("=" * 50)

# 添加新格式

class YAMLFormatter(DataFormatter):

   def format(self, data: dict) -> str:

       lines = []

       for key, value in data.items():

           lines.append(f"{key}: {value}")

       return "\n".join(lines)

# 扩展工厂

FormatterFactory.formatters['yaml'] = YAMLFormatter

# 测试新格式

exporter.set_format('yaml')

result = exporter.export()

print("YAML格式输出:")

print(result)

练习作业

作业1：实现一个ORM元类

创建一个元类，将类属性映射到数据库表的字段。

要求：

自动创建数据库表

支持字段类型验证

实现基本的CRUD操作

支持关系映射（一对一、一对多）

作业2：实现一个属性验证框架

使用描述符创建一个属性验证框架。

要求：

支持多种验证器（长度、范围、正则表达式等）

支持自定义验证函数

支持验证器组合

提供友好的错误信息

作业3：实现性能监控装饰器

创建一个性能监控装饰器。

要求：

记录函数执行时间

记录内存使用变化

统计调用次数

生成性能报告

支持阈值告警

作业4：实现一个缓存系统

创建一个支持多种策略的缓存系统。

要求：

支持LRU、LFU、FIFO缓存策略

支持缓存过期时间

支持缓存统计和监控

支持分布式缓存（选做）

作业5：综合项目 - Web框架

实现一个简易的Web框架。

要求：

支持路由系统

支持中间件

支持模板引擎

支持ORM集成

支持会话管理

学习要点总结

元类（Metaclass）：

元类是创建类的类

可以控制类的创建过程

常用于框架开发和API设计

描述符（Descriptor）：

控制属性访问的协议

实现数据验证、延迟加载等功能

用于构建ORM和验证框架

装饰器高级用法：

参数化装饰器

类装饰器

装饰器堆叠和执行顺序

性能优化：

使用timeit进行基准测试

使用cProfile进行性能分析

掌握常见的性能优化技巧

内存管理：

Python内存管理机制

循环引用和垃圾回收

弱引用的使用

内存分析工具

设计模式：

观察者模式：实现事件驱动架构

策略模式：实现可替换的算法

工厂模式：实现对象创建的封装

常见问题解答

Q: 什么时候应该使用元类？

A: 当需要控制类的创建过程，或者在框架开发中需要自动注册类、验证类定义时。

Q: 描述符和属性的区别是什么？

A: 属性（property）是一种特殊的描述符，描述符更通用，可以控制多个属性的访问。

Q: 如何选择合适的设计模式？

A: 根据问题类型选择：需要解耦观察者和被观察者用观察者模式，需要多种算法用策略模式，需要创建复杂对象用工厂模式。

Q: 如何诊断内存泄漏？

A: 使用memory_profiler监控内存使用，使用objgraph分析对象引用，检查循环引用。

Q: 性能优化的基本原则？

A: 先测量后优化，关注瓶颈，使用合适的数据结构，避免不必要的计算。

下一课预告

第七课将学习：

Python网络编程（socket、HTTP客户端/服务器）

并发编程深入（线程池、进程池、协程池）

数据结构与算法优化

测试与调试技巧

项目结构与代码组织

推荐阅读：

Python中级教程 第六课：高级特性与性能优化 2/2

Python中级教程 第六课：高级特性与性能优化 1/2

Python中级教程 第五课：异步编程与并发 2/2

Python中级教程 第五课：异步编程与并发 1/2

Python中级教程 第四课：上下文管理器、生成器与迭代器

Python中级教程 第三课：面向对象编程深入与装饰器