代理模式

代理模式（Proxy Pattern）为其他对象提供一种代理以控制对这个对象的访问。

问题定义

场景1：延迟加载

# ❌ 问题：立即加载所有资源
class Image:
"""图片类"""
def __init__(self, filename):
self.filename = filename
self._load_image_from_disk()  # 立即加载
def _load_image_from_disk(self):
print(f"从磁盘加载大图片: {self.filename}")
# 模拟加载耗时
time.sleep(1)
def display(self):
print(f"显示图片: {self.filename}")
# 问题：即使不显示，图片也会被加载
image1 = Image("photo1.jpg")  # 立即加载
image2 = Image("photo2.jpg")  # 立即加载
image3 = Image("photo3.jpg")  # 立即加载
# 只有图片1被显示，但3张图都加载了！
image1.display()

场景2：权限控制

# ❌ 问题：没有权限检查
class BankAccount:
"""银行账户"""
def __init__(self, owner, balance):
self.owner = owner
self.balance = balance
def withdraw(self, amount):
"""取款"""
self.balance -= amount
print(f"{self.owner} 取款 ${amount}，余额: ${self.balance}")
# 任何人都可以取款
account = BankAccount("Alice", 1000)
account.withdraw(500)  # 成功
# 但如果Bob想操作Alice的账户呢？
# 没有权限检查！

场景3：远程访问

# ❌ 问题：直接调用远程对象
class RemoteDatabase:
"""远程数据库"""
def query(self, sql):
"""执行查询 - 需要网络通信"""
print(f"通过网络执行查询: {sql}")
# 每次调用都建立连接
return "结果"
# 问题：每次查询都要建立网络连接
db = RemoteDatabase()
db.query("SELECT * FROM users")  # 建立连接
db.query("SELECT * FROM orders")  # 再次建立连接

解决方案

代理模式提供一个代理对象来控制对真实对象的访问。

代理类型

1. 虚拟代理（Virtual Proxy）- 延迟加载

2. 保护代理（Protection Proxy）- 权限控制

3. 远程代理（Remote Proxy）- 远程访问

4. 缓存代理（Cache Proxy）- 缓存结果

标准实现

虚拟代理：延迟加载

import time
class RealImage:
"""真实图片类"""
def __init__(self, filename):
self.filename = filename
self._load_image_from_disk()
def _load_image_from_disk(self):
"""从磁盘加载图片"""
print(f"从磁盘加载大图片: {self.filename}")
time.sleep(1)  # 模拟加载耗时
def display(self):
print(f"显示图片: {self.filename}")
class ImageProxy:
"""图片代理类 - 虚拟代理"""
def __init__(self, filename):
self.filename = filename
self._real_image = None  # 延迟加载
def display(self):
"""显示图片 - 延迟加载"""
if self._real_image is None:
self._real_image = RealImage(self.filename)
self._real_image.display()
# 使用
print("创建图片代理:")
proxy1 = ImageProxy("photo1.jpg")
proxy2 = ImageProxy("photo2.jpg")
proxy3 = ImageProxy("photo3.jpg")
print("\n只有显示时才加载:")
proxy1.display()  # 加载并显示photo1
# 从磁盘加载大图片: photo1.jpg
# 显示图片: photo1.jpg
proxy1.display()  # 不再加载，直接显示
# 显示图片: photo1.jpg
proxy2.display()  # 加载并显示photo2
# 从磁盘加载大图片: photo2.jpg
# 显示图片: photo2.jpg
# photo3从未显示，从未加载！

保护代理：权限控制

from abc import ABC, abstractmethod
class BankAccount(ABC):
"""银行账户接口"""
@abstractmethod
def withdraw(self, amount):
pass
class RealBankAccount(BankAccount):
"""真实银行账户"""
def __init__(self, owner, balance):
self.owner = owner
self.balance = balance
def withdraw(self, amount):
"""取款"""
if amount > self.balance:
raise ValueError("余额不足")
self.balance -= amount
print(f"{self.owner} 取款 ${amount}，余额: ${self.balance}")
class ProtectedBankAccount(BankAccount):
"""保护代理银行账户"""
def __init__(self, real_account, allowed_users):
self._real_account = real_account
self._allowed_users = allowed_users  # 允许操作的用户
def withdraw(self, amount, user):
"""带权限检查的取款"""
if user not in self._allowed_users:
raise PermissionError(f"用户 {user} 没有权限")
print(f"检查权限: {user} ✓")
self._real_account.withdraw(amount)
# 使用
alice_account = RealBankAccount("Alice", 1000)
# 创建保护代理 - 只有Alice可以操作
protected_account = ProtectedBankAccount(alice_account, ["Alice"])
print("Alice 取款:")
protected_account.withdraw(200, "Alice")  # 成功
# 检查权限: Alice ✓
# Alice 取款 $200，余额: $800
print("\nBob 试图取款:")
try:
protected_account.withdraw(100, "Bob")  # 失败
except PermissionError as e:
print(f"错误: {e}")
# 错误: 用户 Bob 没有权限

缓存代理：缓存结果

from abc import ABC, abstractmethod
import time
class DataService(ABC):
"""数据服务接口"""
@abstractmethod
def get_data(self, key):
pass
class RealDataService(DataService):
"""真实数据服务"""
def get_data(self, key):
"""获取数据 - 模拟耗时操作"""
print(f"从数据库查询: {key}")
time.sleep(1)  # 模拟查询耗时
return f"数据-{key}"
class CachedDataService(DataService):
"""缓存数据服务 - 缓存代理"""
def __init__(self, real_service):
self._real_service = real_service
self._cache = {}  # 缓存
def get_data(self, key):
"""获取数据 - 带缓存"""
if key in self._cache:
print(f"从缓存读取: {key}")
return self._cache[key]
# 缓存未命中，查询真实服务
data = self._real_service.get_data(key)
self._cache[key] = data
return data
# 使用
real_service = RealDataService()
cached_service = CachedDataService(real_service)
print("第一次查询:")
data1 = cached_service.get_data("user1")
# 从数据库查询: user1
print("\n第二次查询（相同数据）:")
data2 = cached_service.get_data("user1")
# 从缓存读取: user1
print("\n查询不同数据:")
data3 = cached_service.get_data("user2")
# 从数据库查询: user2
print("\n再次查询相同数据:")
data4 = cached_service.get_data("user1")
# 从缓存读取: user1

实战应用

应用1：网络请求代理

import requests
from abc import ABC, abstractmethod
class APIClient(ABC):
"""API客户端接口"""
@abstractmethod
def get(self, url):
pass
class RealAPIClient(APIClient):
"""真实API客户端"""
def get(self, url):
"""发送HTTP请求"""
print(f"发送请求: {url}")
response = requests.get(url)
return response.json()
class CachedAPIClient(APIClient):
"""缓存API客户端 - 远程+缓存代理"""
def __init__(self, real_client):
self._real_client = real_client
self._cache = {}
def get(self, url):
"""带缓存的请求"""
if url in self._cache:
print(f"从缓存返回: {url}")
return self._cache[url]
# 发送真实请求
response = self._real_client.get(url)
self._cache[url] = response
return response
# 使用
real_api = RealAPIClient()
cached_api = CachedAPIClient(real_api)
# 第一次请求：发送真实请求
# data1 = cached_api.get("https://api.example.com/users")
# 第二次请求：从缓存返回
# data2 = cached_api.get("https://api.example.com/users")

应用2：日志记录代理

from abc import ABC, abstractmethod
class Calculator(ABC):
"""计算器接口"""
@abstractmethod
def add(self, x, y):
pass
class RealCalculator(Calculator):
"""真实计算器"""
def add(self, x, y):
return x + y
class LoggingCalculator(Calculator):
"""日志计算器 - 代理"""
def __init__(self, real_calculator):
self._real_calculator = real_calculator
def add(self, x, y):
"""带日志的计算"""
print(f"[LOG] 执行加法: {x} + {y}")
result = self._real_calculator.add(x, y)
print(f"[LOG] 结果: {result}")
return result
# 使用
real_calc = RealCalculator()
logging_calc = LoggingCalculator(real_calc)
result = logging_calc.add(5, 3)
# [LOG] 执行加法: 5 + 3
# [LOG] 结果: 8

应用3：智能引用代理

class HeavyObject:
"""重对象"""
def __init__(self, data):
self.data = data
print(f"创建重对象，数据: {data}")
class SmartProxy:
"""智能引用代理"""
def __init__(self):
self._object = None
self._reference_count = 0
def acquire(self):
"""获取引用"""
if self._object is None:
self._object = HeavyObject("大数据")
self._reference_count += 1
print(f"引用计数: {self._reference_count}")
return self._object
def release(self):
"""释放引用"""
self._reference_count -= 1
print(f"引用计数: {self._reference_count}")
if self._reference_count == 0:
print("没有引用了，销毁对象")
del self._object
# 使用
proxy = SmartProxy()
obj1 = proxy.acquire()
# 创建重对象，数据: 大数据
# 引用计数: 1
obj2 = proxy.acquire()
# 引用计数: 2
proxy.release()
# 引用计数: 1
proxy.release()
# 引用计数: 0
# 没有引用了，销毁对象

应用4：Python @property 作为代理

class Circle:
"""圆形 - 使用property作为代理"""
def __init__(self, radius):
self._radius = radius  # 私有属性
@property
def radius(self):
"""获取半径"""
print("获取半径")
return self._radius
@radius.setter
def radius(self, value):
"""设置半径 - 带验证"""
if value < 0:
raise ValueError("半径不能为负")
print(f"设置半径为 {value}")
self._radius = value
@property
def area(self):
"""计算面积 - 只读属性"""
print("计算面积")
return 3.14 * self._radius ** 2
# 使用
circle = Circle(5)
# 设置半径为 5
print(circle.radius)
# 获取半径
# 5
print(circle.area)
# 计算面积
# 78.5
circle.radius = 10
# 设置半径为 10

代理 vs 适配器 vs 装饰器

Python内置代理

__getattr__ 和 __getattribute__

class Proxy:
"""通用代理类"""
def __init__(self, target):
self._target = target
def __getattr__(self, name):
"""代理所有未定义的属性访问"""
print(f"代理访问: {name}")
return getattr(self._target, name)
def __setattr__(self, name, value):
"""代理所有属性设置"""
if name == "_target":
super().__setattr__(name, value)
else:
print(f"代理设置: {name} = {value}")
setattr(self._target, name, value)
# 使用
class Target:
def __init__(self):
self.value = 10
def method(self):
return "执行方法"
target = Target()
proxy = Proxy(target)
print(proxy.value)
# 代理访问: value
# 10
proxy.method()
# 代理访问: method
# 执行方法
proxy.value = 20
# 代理设置: value = 20

优缺点

✅ 优点

❌ 缺点

本章要点

• ✅ 代理模式控制对对象的访问
• ✅ 虚拟代理：延迟加载
• ✅ 保护代理：权限控制
• ✅ 缓存代理：缓存结果
• ✅ 远程代理：网络访问
• ✅ Python的@property和__getattr__可用于实现代理
• ✅ 与适配器、装饰器模式目的不同

下一步：外观模式 🚀