6.1 外观模式（Facade Pattern）
6.1.1 模式定义与动机
定义： 外观模式为子系统中的一组接口提供一个一致的界面，外观模式定义了一个高层接口，这个接口使得这一子系统更加容易使用。
动机： - 子系统越来越复杂，需要简化客户端的使用 - 减少客户端与子系统之间的依赖关系 - 为复杂的子系统提供简单的接口 - 实现子系统的分层
6.1.2 模式结构
┌─────────────┐ ┌─────────────┐
│ Client │───▶│ Facade │
│ │ │ │
└─────────────┘ └─────────────┘
│
▼
┌─────────────┐
│ Subsystem │
│ Classes │
│ A, B, C... │
└─────────────┘
角色说明： - Facade（外观）： 知道哪些子系统类负责处理请求，将客户的请求代理给适当的子系统对象 - Subsystem Classes（子系统类）： 实现子系统的功能，处理由Facade对象指派的任务 - Client（客户端）： 通过Facade接口与子系统通信
6.1.3 Python实现示例
# Python外观模式实现 - 智能家居系统
from abc import ABC, abstractmethod
from typing import Dict, List, Any
from enum import Enum
import time
from datetime import datetime
# 子系统1：照明系统
class LightingSystem:
"""照明子系统"""
def __init__(self):
self._lights = {
'living_room': {'brightness': 0, 'color': 'white', 'on': False},
'bedroom': {'brightness': 0, 'color': 'white', 'on': False},
'kitchen': {'brightness': 0, 'color': 'white', 'on': False},
'bathroom': {'brightness': 0, 'color': 'white', 'on': False}
}
def turn_on_light(self, room: str, brightness: int = 100, color: str = 'white'):
"""开启指定房间的灯光"""
if room in self._lights:
self._lights[room] = {
'brightness': brightness,
'color': color,
'on': True
}
print(f"[LightingSystem] {room} light turned ON - Brightness: {brightness}%, Color: {color}")
else:
print(f"[LightingSystem] Room '{room}' not found")
def turn_off_light(self, room: str):
"""关闭指定房间的灯光"""
if room in self._lights:
self._lights[room]['on'] = False
self._lights[room]['brightness'] = 0
print(f"[LightingSystem] {room} light turned OFF")
else:
print(f"[LightingSystem] Room '{room}' not found")
def dim_light(self, room: str, brightness: int):
"""调节灯光亮度"""
if room in self._lights and self._lights[room]['on']:
self._lights[room]['brightness'] = max(0, min(100, brightness))
print(f"[LightingSystem] {room} light dimmed to {brightness}%")
else:
print(f"[LightingSystem] Cannot dim light in '{room}' - light is off or room not found")
def turn_on_all_lights(self, brightness: int = 80):
"""开启所有灯光"""
for room in self._lights:
self.turn_on_light(room, brightness)
def turn_off_all_lights(self):
"""关闭所有灯光"""
for room in self._lights:
self.turn_off_light(room)
def get_status(self) -> Dict[str, Any]:
"""获取照明系统状态"""
return self._lights.copy()
# 子系统2：安全系统
class SecuritySystem:
"""安全子系统"""
def __init__(self):
self._armed = False
self._alarm_triggered = False
self._sensors = {
'door': {'active': True, 'triggered': False},
'window': {'active': True, 'triggered': False},
'motion': {'active': True, 'triggered': False},
'smoke': {'active': True, 'triggered': False}
}
self._access_code = "1234"
def arm_system(self, code: str):
"""布防安全系统"""
if code == self._access_code:
self._armed = True
self._alarm_triggered = False
print("[SecuritySystem] Security system ARMED")
return True
else:
print("[SecuritySystem] Invalid access code - cannot arm system")
return False
def disarm_system(self, code: str):
"""撤防安全系统"""
if code == self._access_code:
self._armed = False
self._alarm_triggered = False
print("[SecuritySystem] Security system DISARMED")
return True
else:
print("[SecuritySystem] Invalid access code - cannot disarm system")
return False
def trigger_alarm(self, sensor_type: str):
"""触发警报"""
if self._armed and sensor_type in self._sensors:
self._alarm_triggered = True
self._sensors[sensor_type]['triggered'] = True
print(f"[SecuritySystem] ALARM TRIGGERED by {sensor_type} sensor!")
def reset_alarm(self, code: str):
"""重置警报"""
if code == self._access_code:
self._alarm_triggered = False
for sensor in self._sensors.values():
sensor['triggered'] = False
print("[SecuritySystem] Alarm reset")
return True
else:
print("[SecuritySystem] Invalid access code - cannot reset alarm")
return False
def get_status(self) -> Dict[str, Any]:
"""获取安全系统状态"""
return {
'armed': self._armed,
'alarm_triggered': self._alarm_triggered,
'sensors': self._sensors.copy()
}
# 子系统3：气候控制系统
class ClimateControlSystem:
"""气候控制子系统"""
def __init__(self):
self._temperature = 22 # 摄氏度
self._humidity = 50 # 百分比
self._ac_on = False
self._heater_on = False
self._fan_speed = 0 # 0-5级
self._target_temperature = 22
def set_temperature(self, temperature: int):
"""设置目标温度"""
self._target_temperature = max(16, min(30, temperature))
self._adjust_climate()
print(f"[ClimateControl] Target temperature set to {self._target_temperature}°C")
def turn_on_ac(self, temperature: int = 24):
"""开启空调"""
self._ac_on = True
self._heater_on = False
self._target_temperature = temperature
self._fan_speed = 3
print(f"[ClimateControl] Air conditioning turned ON - Target: {temperature}°C")
def turn_on_heater(self, temperature: int = 22):
"""开启暖气"""
self._heater_on = True
self._ac_on = False
self._target_temperature = temperature
self._fan_speed = 2
print(f"[ClimateControl] Heater turned ON - Target: {temperature}°C")
def turn_off_climate_control(self):
"""关闭气候控制"""
self._ac_on = False
self._heater_on = False
self._fan_speed = 0
print("[ClimateControl] Climate control turned OFF")
def set_fan_speed(self, speed: int):
"""设置风扇速度"""
self._fan_speed = max(0, min(5, speed))
print(f"[ClimateControl] Fan speed set to {self._fan_speed}")
def _adjust_climate(self):
"""自动调节气候"""
if self._target_temperature > self._temperature:
if not self._heater_on:
self.turn_on_heater(self._target_temperature)
elif self._target_temperature < self._temperature:
if not self._ac_on:
self.turn_on_ac(self._target_temperature)
def get_status(self) -> Dict[str, Any]:
"""获取气候控制状态"""
return {
'current_temperature': self._temperature,
'target_temperature': self._target_temperature,
'humidity': self._humidity,
'ac_on': self._ac_on,
'heater_on': self._heater_on,
'fan_speed': self._fan_speed
}
# 子系统4：娱乐系统
class EntertainmentSystem:
"""娱乐子系统"""
def __init__(self):
self._tv_on = False
self._tv_channel = 1
self._tv_volume = 20
self._sound_system_on = False
self._sound_volume = 30
self._music_playing = False
self._current_song = None
def turn_on_tv(self, channel: int = 1, volume: int = 20):
"""开启电视"""
self._tv_on = True
self._tv_channel = channel
self._tv_volume = volume
print(f"[Entertainment] TV turned ON - Channel: {channel}, Volume: {volume}")
def turn_off_tv(self):
"""关闭电视"""
self._tv_on = False
print("[Entertainment] TV turned OFF")
def change_channel(self, channel: int):
"""换台"""
if self._tv_on:
self._tv_channel = channel
print(f"[Entertainment] TV channel changed to {channel}")
else:
print("[Entertainment] Cannot change channel - TV is off")
def set_tv_volume(self, volume: int):
"""设置电视音量"""
if self._tv_on:
self._tv_volume = max(0, min(100, volume))
print(f"[Entertainment] TV volume set to {self._tv_volume}")
def turn_on_sound_system(self, volume: int = 30):
"""开启音响系统"""
self._sound_system_on = True
self._sound_volume = volume
print(f"[Entertainment] Sound system turned ON - Volume: {volume}")
def turn_off_sound_system(self):
"""关闭音响系统"""
self._sound_system_on = False
self._music_playing = False
print("[Entertainment] Sound system turned OFF")
def play_music(self, song: str = "Default Playlist"):
"""播放音乐"""
if self._sound_system_on:
self._music_playing = True
self._current_song = song
print(f"[Entertainment] Playing music: {song}")
else:
print("[Entertainment] Cannot play music - sound system is off")
def stop_music(self):
"""停止音乐"""
self._music_playing = False
self._current_song = None
print("[Entertainment] Music stopped")
def get_status(self) -> Dict[str, Any]:
"""获取娱乐系统状态"""
return {
'tv_on': self._tv_on,
'tv_channel': self._tv_channel,
'tv_volume': self._tv_volume,
'sound_system_on': self._sound_system_on,
'sound_volume': self._sound_volume,
'music_playing': self._music_playing,
'current_song': self._current_song
}
# 外观类：智能家居控制器
class SmartHomeFacade:
"""智能家居外观类"""
def __init__(self):
# 初始化所有子系统
self._lighting = LightingSystem()
self._security = SecuritySystem()
self._climate = ClimateControlSystem()
self._entertainment = EntertainmentSystem()
# 预定义场景
self._scenes = {
'morning': self._morning_scene,
'evening': self._evening_scene,
'night': self._night_scene,
'away': self._away_scene,
'party': self._party_scene,
'movie': self._movie_scene
}
# 高级场景控制方法
def activate_scene(self, scene_name: str, **kwargs):
"""激活预定义场景"""
if scene_name in self._scenes:
print(f"\n=== Activating {scene_name.upper()} Scene ===")
self._scenes[scene_name](**kwargs)
print(f"=== {scene_name.upper()} Scene Activated ===\n")
else:
print(f"Unknown scene: {scene_name}")
print(f"Available scenes: {list(self._scenes.keys())}")
def _morning_scene(self, **kwargs):
"""晨起场景"""
# 逐渐开启灯光
self._lighting.turn_on_light('bedroom', 30, 'warm_white')
self._lighting.turn_on_light('bathroom', 80, 'white')
self._lighting.turn_on_light('kitchen', 90, 'white')
# 设置舒适温度
self._climate.set_temperature(23)
# 播放轻音乐
self._entertainment.turn_on_sound_system(20)
self._entertainment.play_music("Morning Playlist")
# 撤防安全系统
code = kwargs.get('security_code', '1234')
self._security.disarm_system(code)
def _evening_scene(self, **kwargs):
"""傍晚场景"""
# 开启温馨灯光
self._lighting.turn_on_light('living_room', 60, 'warm_white')
self._lighting.turn_on_light('kitchen', 70, 'white')
self._lighting.turn_on_light('bedroom', 40, 'warm_white')
# 调节温度
self._climate.set_temperature(22)
# 开启电视
self._entertainment.turn_on_tv(5, 25)
def _night_scene(self, **kwargs):
"""夜间场景"""
# 关闭大部分灯光，保留夜灯
self._lighting.turn_off_all_lights()
self._lighting.turn_on_light('bedroom', 10, 'blue')
self._lighting.turn_on_light('bathroom', 20, 'warm_white')
# 降低温度
self._climate.set_temperature(20)
# 关闭娱乐设备
self._entertainment.turn_off_tv()
self._entertainment.turn_off_sound_system()
# 布防安全系统
code = kwargs.get('security_code', '1234')
self._security.arm_system(code)
def _away_scene(self, **kwargs):
"""离家场景"""
# 关闭所有灯光
self._lighting.turn_off_all_lights()
# 关闭气候控制
self._climate.turn_off_climate_control()
# 关闭所有娱乐设备
self._entertainment.turn_off_tv()
self._entertainment.turn_off_sound_system()
# 布防安全系统
code = kwargs.get('security_code', '1234')
self._security.arm_system(code)
def _party_scene(self, **kwargs):
"""聚会场景"""
# 开启彩色灯光
self._lighting.turn_on_light('living_room', 100, 'rainbow')
self._lighting.turn_on_light('kitchen', 80, 'white')
# 设置舒适温度
self._climate.set_temperature(21)
# 开启音响系统
self._entertainment.turn_on_sound_system(60)
self._entertainment.play_music("Party Playlist")
# 撤防安全系统（客人进出）
code = kwargs.get('security_code', '1234')
self._security.disarm_system(code)
def _movie_scene(self, **kwargs):
"""观影场景"""
# 调暗灯光
self._lighting.turn_off_all_lights()
self._lighting.turn_on_light('living_room', 20, 'blue')
# 设置舒适温度
self._climate.set_temperature(22)
# 开启电视和音响
self._entertainment.turn_on_tv(kwargs.get('channel', 10), 30)
self._entertainment.turn_on_sound_system(50)
# 简化的控制方法
def good_morning(self, security_code: str = '1234'):
"""早安模式"""
self.activate_scene('morning', security_code=security_code)
def good_night(self, security_code: str = '1234'):
"""晚安模式"""
self.activate_scene('night', security_code=security_code)
def leaving_home(self, security_code: str = '1234'):
"""离家模式"""
self.activate_scene('away', security_code=security_code)
def coming_home(self, security_code: str = '1234'):
"""回家模式"""
self.activate_scene('evening', security_code=security_code)
def start_party(self, security_code: str = '1234'):
"""聚会模式"""
self.activate_scene('party', security_code=security_code)
def watch_movie(self, channel: int = 10, security_code: str = '1234'):
"""观影模式"""
self.activate_scene('movie', channel=channel, security_code=security_code)
# 紧急控制方法
def emergency_shutdown(self):
"""紧急关闭所有系统"""
print("\n!!! EMERGENCY SHUTDOWN !!!")
self._lighting.turn_off_all_lights()
self._climate.turn_off_climate_control()
self._entertainment.turn_off_tv()
self._entertainment.turn_off_sound_system()
print("All systems shut down for safety")
def security_alert(self, sensor_type: str = 'motion'):
"""安全警报"""
print("\n!!! SECURITY ALERT !!!")
self._security.trigger_alarm(sensor_type)
# 开启所有灯光
self._lighting.turn_on_all_lights(100)
# 关闭娱乐系统
self._entertainment.turn_off_tv()
self._entertainment.turn_off_sound_system()
# 状态查询方法
def get_system_status(self) -> Dict[str, Any]:
"""获取整个系统状态"""
return {
'timestamp': datetime.now().isoformat(),
'lighting': self._lighting.get_status(),
'security': self._security.get_status(),
'climate': self._climate.get_status(),
'entertainment': self._entertainment.get_status()
}
def get_available_scenes(self) -> List[str]:
"""获取可用场景列表"""
return list(self._scenes.keys())
# 直接访问子系统的方法（可选）
def get_lighting_system(self) -> LightingSystem:
"""获取照明系统（用于高级控制）"""
return self._lighting
def get_security_system(self) -> SecuritySystem:
"""获取安全系统（用于高级控制）"""
return self._security
def get_climate_system(self) -> ClimateControlSystem:
"""获取气候系统（用于高级控制）"""
return self._climate
def get_entertainment_system(self) -> EntertainmentSystem:
"""获取娱乐系统（用于高级控制）"""
return self._entertainment
# 使用示例和演示
def demonstrate_facade_pattern():
print("=== Facade Pattern Demo - Smart Home System ===")
print()
# 创建智能家居控制器
smart_home = SmartHomeFacade()
print(f"Available scenes: {smart_home.get_available_scenes()}")
print()
# 演示1：早晨场景
print("1. Morning Routine:")
smart_home.good_morning()
# 演示2：查看系统状态
print("2. System Status Check:")
status = smart_home.get_system_status()
print(f"Lighting status: {len([room for room, info in status['lighting'].items() if info['on']])} rooms lit")
print(f"Security armed: {status['security']['armed']}")
print(f"Climate target: {status['climate']['target_temperature']}°C")
print(f"Entertainment: TV={status['entertainment']['tv_on']}, Music={status['entertainment']['music_playing']}")
print()
# 演示3：观影模式
print("3. Movie Night:")
smart_home.watch_movie(channel=15)
# 演示4：聚会模式
print("4. Party Time:")
smart_home.start_party()
# 演示5：离家模式
print("5. Leaving Home:")
smart_home.leaving_home()
# 演示6：安全警报
print("6. Security Alert Simulation:")
smart_home.security_alert('door')
# 演示7：紧急关闭
print("7. Emergency Shutdown:")
smart_home.emergency_shutdown()
# 演示8：高级控制（直接访问子系统）
print("8. Advanced Control (Direct Subsystem Access):")
lighting = smart_home.get_lighting_system()
lighting.turn_on_light('living_room', 50, 'purple')
climate = smart_home.get_climate_system()
climate.set_fan_speed(4)
print("\n" + "="*60)
print("Facade Pattern Benefits Demonstrated:")
print("1. Simplified interface for complex subsystems")
print("2. Reduced coupling between client and subsystems")
print("3. Layered architecture with high-level operations")
print("4. Easy-to-use predefined scenarios")
print("5. Optional direct access to subsystems for advanced users")
if __name__ == "__main__":
demonstrate_facade_pattern()
6.1.4 外观模式的优缺点
优点： 1. 简化接口： 为复杂子系统提供简单接口 2. 降低耦合： 减少客户端与子系统的依赖 3. 分层设计： 支持系统的分层架构 4. 易于使用： 客户端无需了解子系统的复杂性
缺点： 1. 功能限制： 可能无法提供子系统的所有功能 2. 违反开闭原则： 增加新的子系统可能需要修改外观类 3. 性能开销： 增加了一层间接调用
6.1.5 适用场景
- 复杂子系统简化： 当子系统很复杂时
- 分层架构： 构建分层系统时的层间接口
- 遗留系统集成： 为老系统提供现代化接口
- API设计： 为复杂的API提供简化版本
6.2 代理模式（Proxy Pattern）
6.2.1 模式定义与动机
定义： 代理模式为其他对象提供一种代理以控制对这个对象的访问。
动机： - 控制对象的访问权限 - 延迟对象的创建和初始化 - 在访问对象时执行额外的操作 - 为远程对象提供本地代表
6.2.2 代理模式的类型
- 虚拟代理（Virtual Proxy）： 延迟创建开销大的对象
- 保护代理（Protection Proxy）： 控制对原始对象的访问权限
- 远程代理（Remote Proxy）： 为远程对象提供本地代理
- 缓存代理（Cache Proxy）： 为开销大的运算结果提供缓存
- 智能引用代理（Smart Reference Proxy）： 提供额外的服务
6.2.3 模式结构
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Client │───▶│ Subject │◀───│ RealSubject │
│ │ │ (interface) │ │ │
└─────────────┘ └─────────────┘ └─────────────┘
▲
│
┌─────────────┐
│ Proxy │
│ │
└─────────────┘
6.2.4 Python实现示例
# Python代理模式实现 - 图像加载系统
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional, List
import time
import hashlib
from datetime import datetime, timedelta
from enum import Enum
# 权限枚举
class Permission(Enum):
READ = "read"
WRITE = "write"
DELETE = "delete"
ADMIN = "admin"
# 用户类
class User:
"""用户类"""
def __init__(self, username: str, permissions: List[Permission]):
self.username = username
self.permissions = permissions
self.login_time = datetime.now()
def has_permission(self, permission: Permission) -> bool:
return permission in self.permissions or Permission.ADMIN in self.permissions
# 抽象主题接口
class ImageService(ABC):
"""图像服务接口"""
@abstractmethod
def load_image(self, image_path: str) -> Dict[str, Any]:
"""加载图像"""
pass
@abstractmethod
def save_image(self, image_path: str, image_data: bytes) -> bool:
"""保存图像"""
pass
@abstractmethod
def delete_image(self, image_path: str) -> bool:
"""删除图像"""
pass
@abstractmethod
def get_image_info(self, image_path: str) -> Dict[str, Any]:
"""获取图像信息"""
pass
# 真实主题：实际的图像服务
class RealImageService(ImageService):
"""真实的图像服务实现"""
def __init__(self):
self._images = {} # 模拟图像存储
self._load_count = 0
self._save_count = 0
self._delete_count = 0
def load_image(self, image_path: str) -> Dict[str, Any]:
"""加载图像（模拟耗时操作）"""
print(f"[RealImageService] Loading image: {image_path}")
# 模拟图像加载的耗时操作
time.sleep(0.5) # 模拟I/O延迟
self._load_count += 1
if image_path in self._images:
image_data = self._images[image_path]
return {
'status': 'success',
'path': image_path,
'size': image_data['size'],
'format': image_data['format'],
'data': image_data['data'],
'loaded_at': datetime.now().isoformat(),
'load_time': 0.5
}
else:
return {
'status': 'not_found',
'path': image_path,
'error': 'Image not found'
}
def save_image(self, image_path: str, image_data: bytes) -> bool:
"""保存图像"""
print(f"[RealImageService] Saving image: {image_path}")
# 模拟图像保存操作
time.sleep(0.2)
self._save_count += 1
# 模拟图像元数据
self._images[image_path] = {
'data': image_data,
'size': len(image_data),
'format': self._detect_format(image_path),
'created_at': datetime.now().isoformat(),
'modified_at': datetime.now().isoformat()
}
return True
def delete_image(self, image_path: str) -> bool:
"""删除图像"""
print(f"[RealImageService] Deleting image: {image_path}")
self._delete_count += 1
if image_path in self._images:
del self._images[image_path]
return True
else:
return False
def get_image_info(self, image_path: str) -> Dict[str, Any]:
"""获取图像信息"""
if image_path in self._images:
return self._images[image_path].copy()
else:
return {'error': 'Image not found'}
def _detect_format(self, image_path: str) -> str:
"""检测图像格式"""
if image_path.lower().endswith('.jpg') or image_path.lower().endswith('.jpeg'):
return 'JPEG'
elif image_path.lower().endswith('.png'):
return 'PNG'
elif image_path.lower().endswith('.gif'):
return 'GIF'
else:
return 'UNKNOWN'
def get_statistics(self) -> Dict[str, int]:
"""获取服务统计信息"""
return {
'load_count': self._load_count,
'save_count': self._save_count,
'delete_count': self._delete_count,
'total_images': len(self._images)
}
# 虚拟代理：延迟加载
class VirtualImageProxy(ImageService):
"""虚拟代理 - 延迟加载图像服务"""
def __init__(self):
self._real_service: Optional[RealImageService] = None
self._initialized = False
def _get_real_service(self) -> RealImageService:
"""延迟初始化真实服务"""
if self._real_service is None:
print("[VirtualProxy] Initializing real image service...")
self._real_service = RealImageService()
self._initialized = True
return self._real_service
def load_image(self, image_path: str) -> Dict[str, Any]:
print(f"[VirtualProxy] Request to load image: {image_path}")
real_service = self._get_real_service()
return real_service.load_image(image_path)
def save_image(self, image_path: str, image_data: bytes) -> bool:
print(f"[VirtualProxy] Request to save image: {image_path}")
real_service = self._get_real_service()
return real_service.save_image(image_path, image_data)
def delete_image(self, image_path: str) -> bool:
print(f"[VirtualProxy] Request to delete image: {image_path}")
real_service = self._get_real_service()
return real_service.delete_image(image_path)
def get_image_info(self, image_path: str) -> Dict[str, Any]:
print(f"[VirtualProxy] Request for image info: {image_path}")
real_service = self._get_real_service()
return real_service.get_image_info(image_path)
def is_initialized(self) -> bool:
"""检查是否已初始化"""
return self._initialized
# 保护代理：权限控制
class ProtectionImageProxy(ImageService):
"""保护代理 - 权限控制"""
def __init__(self, real_service: ImageService, current_user: User):
self._real_service = real_service
self._current_user = current_user
self._access_log = []
def load_image(self, image_path: str) -> Dict[str, Any]:
if self._check_permission(Permission.READ, 'load_image', image_path):
return self._real_service.load_image(image_path)
else:
return {
'status': 'access_denied',
'error': 'Insufficient permissions to load image'
}
def save_image(self, image_path: str, image_data: bytes) -> bool:
if self._check_permission(Permission.WRITE, 'save_image', image_path):
return self._real_service.save_image(image_path, image_data)
else:
print(f"[ProtectionProxy] Access denied: {self._current_user.username} cannot save images")
return False
def delete_image(self, image_path: str) -> bool:
if self._check_permission(Permission.DELETE, 'delete_image', image_path):
return self._real_service.delete_image(image_path)
else:
print(f"[ProtectionProxy] Access denied: {self._current_user.username} cannot delete images")
return False
def get_image_info(self, image_path: str) -> Dict[str, Any]:
if self._check_permission(Permission.READ, 'get_image_info', image_path):
return self._real_service.get_image_info(image_path)
else:
return {
'error': 'Insufficient permissions to access image info'
}
def _check_permission(self, required_permission: Permission, operation: str, resource: str) -> bool:
"""检查权限"""
has_permission = self._current_user.has_permission(required_permission)
# 记录访问日志
self._access_log.append({
'timestamp': datetime.now().isoformat(),
'user': self._current_user.username,
'operation': operation,
'resource': resource,
'permission_required': required_permission.value,
'access_granted': has_permission
})
if has_permission:
print(f"[ProtectionProxy] Access granted: {self._current_user.username} -> {operation}")
else:
print(f"[ProtectionProxy] Access denied: {self._current_user.username} -> {operation} (requires {required_permission.value})")
return has_permission
def get_access_log(self) -> List[Dict[str, Any]]:
"""获取访问日志"""
return self._access_log.copy()
# 缓存代理：结果缓存
class CacheImageProxy(ImageService):
"""缓存代理 - 缓存图像加载结果"""
def __init__(self, real_service: ImageService, cache_ttl: int = 300):
self._real_service = real_service
self._cache = {}
self._cache_ttl = cache_ttl # 缓存生存时间（秒）
self._cache_hits = 0
self._cache_misses = 0
def load_image(self, image_path: str) -> Dict[str, Any]:
cache_key = self._generate_cache_key('load', image_path)
# 检查缓存
cached_result = self._get_from_cache(cache_key)
if cached_result is not None:
print(f"[CacheProxy] Cache hit for image: {image_path}")
self._cache_hits += 1
cached_result['cache_hit'] = True
return cached_result
print(f"[CacheProxy] Cache miss for image: {image_path}")
self._cache_misses += 1
# 缓存未命中，调用真实服务
result = self._real_service.load_image(image_path)
result['cache_hit'] = False
# 只缓存成功的结果
if result.get('status') == 'success':
self._add_to_cache(cache_key, result)
return result
def save_image(self, image_path: str, image_data: bytes) -> bool:
# 保存操作会使相关缓存失效
self._invalidate_cache_for_image(image_path)
return self._real_service.save_image(image_path, image_data)
def delete_image(self, image_path: str) -> bool:
# 删除操作会使相关缓存失效
self._invalidate_cache_for_image(image_path)
return self._real_service.delete_image(image_path)
def get_image_info(self, image_path: str) -> Dict[str, Any]:
cache_key = self._generate_cache_key('info', image_path)
# 检查缓存
cached_result = self._get_from_cache(cache_key)
if cached_result is not None:
print(f"[CacheProxy] Cache hit for image info: {image_path}")
self._cache_hits += 1
return cached_result
print(f"[CacheProxy] Cache miss for image info: {image_path}")
self._cache_misses += 1
# 缓存未命中，调用真实服务
result = self._real_service.get_image_info(image_path)
# 只缓存成功的结果
if 'error' not in result:
self._add_to_cache(cache_key, result)
return result
def _generate_cache_key(self, operation: str, image_path: str) -> str:
"""生成缓存键"""
key_string = f"{operation}:{image_path}"
return hashlib.md5(key_string.encode()).hexdigest()
def _get_from_cache(self, cache_key: str) -> Optional[Dict[str, Any]]:
"""从缓存获取数据"""
if cache_key in self._cache:
cache_entry = self._cache[cache_key]
if time.time() - cache_entry['timestamp'] < self._cache_ttl:
return cache_entry['data'].copy()
else:
# 缓存过期，删除
del self._cache[cache_key]
return None
def _add_to_cache(self, cache_key: str, data: Dict[str, Any]):
"""添加到缓存"""
self._cache[cache_key] = {
'data': data.copy(),
'timestamp': time.time()
}
def _invalidate_cache_for_image(self, image_path: str):
"""使指定图像的缓存失效"""
keys_to_remove = []
for cache_key, cache_entry in self._cache.items():
# 检查缓存键是否与图像路径相关
if image_path in cache_key:
keys_to_remove.append(cache_key)
for key in keys_to_remove:
del self._cache[key]
print(f"[CacheProxy] Invalidated cache for key: {key}")
def get_cache_statistics(self) -> Dict[str, Any]:
"""获取缓存统计信息"""
total_requests = self._cache_hits + self._cache_misses
hit_rate = (self._cache_hits / total_requests * 100) if total_requests > 0 else 0
return {
'cache_hits': self._cache_hits,
'cache_misses': self._cache_misses,
'hit_rate': f"{hit_rate:.2f}%",
'cache_size': len(self._cache),
'cache_ttl': self._cache_ttl
}
# 智能引用代理：额外服务
class SmartReferenceProxy(ImageService):
"""智能引用代理 - 提供额外的服务和统计"""
def __init__(self, real_service: ImageService):
self._real_service = real_service
self._reference_count = {}
self._operation_history = []
self._performance_metrics = {
'total_operations': 0,
'total_time': 0.0,
'average_time': 0.0
}
def load_image(self, image_path: str) -> Dict[str, Any]:
return self._execute_with_tracking('load_image', image_path,
lambda: self._real_service.load_image(image_path))
def save_image(self, image_path: str, image_data: bytes) -> bool:
return self._execute_with_tracking('save_image', image_path,
lambda: self._real_service.save_image(image_path, image_data))
def delete_image(self, image_path: str) -> bool:
result = self._execute_with_tracking('delete_image', image_path,
lambda: self._real_service.delete_image(image_path))
# 删除引用计数
if result and image_path in self._reference_count:
del self._reference_count[image_path]
return result
def get_image_info(self, image_path: str) -> Dict[str, Any]:
return self._execute_with_tracking('get_image_info', image_path,
lambda: self._real_service.get_image_info(image_path))
def _execute_with_tracking(self, operation: str, image_path: str, func):
"""执行操作并跟踪统计信息"""
start_time = time.time()
# 更新引用计数
if image_path not in self._reference_count:
self._reference_count[image_path] = 0
self._reference_count[image_path] += 1
try:
result = func()
execution_time = time.time() - start_time
# 记录操作历史
self._operation_history.append({
'timestamp': datetime.now().isoformat(),
'operation': operation,
'image_path': image_path,
'execution_time': execution_time,
'success': True
})
# 更新性能指标
self._update_performance_metrics(execution_time)
print(f"[SmartProxy] {operation} completed in {execution_time:.3f}s")
return result
except Exception as e:
execution_time = time.time() - start_time
# 记录失败的操作
self._operation_history.append({
'timestamp': datetime.now().isoformat(),
'operation': operation,
'image_path': image_path,
'execution_time': execution_time,
'success': False,
'error': str(e)
})
print(f"[SmartProxy] {operation} failed after {execution_time:.3f}s: {str(e)}")
raise
def _update_performance_metrics(self, execution_time: float):
"""更新性能指标"""
self._performance_metrics['total_operations'] += 1
self._performance_metrics['total_time'] += execution_time
self._performance_metrics['average_time'] = (
self._performance_metrics['total_time'] / self._performance_metrics['total_operations']
)
def get_reference_count(self, image_path: str) -> int:
"""获取图像的引用计数"""
return self._reference_count.get(image_path, 0)
def get_all_reference_counts(self) -> Dict[str, int]:
"""获取所有图像的引用计数"""
return self._reference_count.copy()
def get_operation_history(self, limit: int = 10) -> List[Dict[str, Any]]:
"""获取操作历史"""
return self._operation_history[-limit:]
def get_performance_metrics(self) -> Dict[str, Any]:
"""获取性能指标"""
return self._performance_metrics.copy()
def get_most_accessed_images(self, top_n: int = 5) -> List[tuple]:
"""获取访问次数最多的图像"""
sorted_images = sorted(self._reference_count.items(),
key=lambda x: x[1], reverse=True)
return sorted_images[:top_n]
# 使用示例和演示
def demonstrate_proxy_pattern():
print("=== Proxy Pattern Demo ===")
print()
# 创建用户
admin_user = User("admin", [Permission.ADMIN])
read_user = User("reader", [Permission.READ])
write_user = User("writer", [Permission.READ, Permission.WRITE])
print("1. Virtual Proxy Demo (Lazy Initialization):")
print("-" * 50)
# 虚拟代理演示
virtual_proxy = VirtualImageProxy()
print(f"Proxy initialized: {virtual_proxy.is_initialized()}")
# 第一次访问会触发真实服务的初始化
virtual_proxy.save_image("test1.jpg", b"fake_image_data_1")
print(f"Proxy initialized after first use: {virtual_proxy.is_initialized()}")
print()
print("2. Protection Proxy Demo (Access Control):")
print("-" * 50)
# 创建真实服务并添加一些测试数据
real_service = RealImageService()
real_service.save_image("public.jpg", b"public_image_data")
real_service.save_image("private.jpg", b"private_image_data")
# 保护代理演示 - 只读用户
protection_proxy_read = ProtectionImageProxy(real_service, read_user)
print("Testing with read-only user:")
result = protection_proxy_read.load_image("public.jpg")
print(f"Load result: {result.get('status')}")
success = protection_proxy_read.save_image("new.jpg", b"new_data")
print(f"Save result: {success}")
success = protection_proxy_read.delete_image("public.jpg")
print(f"Delete result: {success}")
print("\nTesting with write user:")
protection_proxy_write = ProtectionImageProxy(real_service, write_user)
success = protection_proxy_write.save_image("new.jpg", b"new_data")
print(f"Save result: {success}")
print("\nAccess log:")
for log_entry in protection_proxy_read.get_access_log()[-3:]:
print(f" {log_entry['operation']}: {log_entry['access_granted']}")
print()
print("3. Cache Proxy Demo (Result Caching):")
print("-" * 50)
# 缓存代理演示
cache_proxy = CacheImageProxy(real_service, cache_ttl=60)
# 第一次加载（缓存未命中）
print("First load (cache miss):")
result1 = cache_proxy.load_image("public.jpg")
print(f"Cache hit: {result1.get('cache_hit')}")
# 第二次加载（缓存命中）
print("\nSecond load (cache hit):")
result2 = cache_proxy.load_image("public.jpg")
print(f"Cache hit: {result2.get('cache_hit')}")
print(f"\nCache statistics: {cache_proxy.get_cache_statistics()}")
print()
print("4. Smart Reference Proxy Demo (Additional Services):")
print("-" * 50)
# 智能引用代理演示
smart_proxy = SmartReferenceProxy(real_service)
# 执行一些操作
smart_proxy.load_image("public.jpg")
smart_proxy.load_image("public.jpg") # 再次访问
smart_proxy.load_image("private.jpg")
smart_proxy.get_image_info("public.jpg")
print(f"\nReference counts: {smart_proxy.get_all_reference_counts()}")
print(f"Performance metrics: {smart_proxy.get_performance_metrics()}")
print(f"Most accessed images: {smart_proxy.get_most_accessed_images()}")
print("\nOperation history:")
for operation in smart_proxy.get_operation_history(3):
print(f" {operation['operation']}: {operation['execution_time']:.3f}s")
print("\n" + "="*60)
print("Proxy Pattern Benefits Demonstrated:")
print("1. Virtual Proxy: Lazy initialization of expensive objects")
print("2. Protection Proxy: Access control and security")
print("3. Cache Proxy: Performance optimization through caching")
print("4. Smart Reference: Additional services and monitoring")
print("5. Transparent interface: Client code remains unchanged")
if __name__ == "__main__":
demonstrate_proxy_pattern()
6.2.5 代理模式的优缺点
优点： 1. 职责分离： 代理和真实对象分离关注点 2. 透明性： 客户端无需知道是否使用了代理 3. 灵活性： 可以在运行时决定是否使用代理 4. 扩展性： 可以在不修改真实对象的情况下扩展功能
缺点： 1. 复杂性增加： 增加了系统的复杂性 2. 性能开销： 增加了间接访问的开销 3. 维护成本： 需要维护代理类
6.2.6 适用场景
- 延迟加载： 对象创建开销大时
- 访问控制： 需要控制对象访问权限时
- 缓存： 需要缓存昂贵操作的结果时
- 远程访问： 访问远程对象时
- 引用计数： 需要跟踪对象使用情况时
6.3 外观模式与代理模式对比
| 特性 | 外观模式 | 代理模式 |
|---|---|---|
| 目的 | 简化复杂接口 | 控制对象访问 |
| 结构 | 一对多（外观对多个子系统） | 一对一（代理对真实对象） |
| 接口 | 提供新的简化接口 | 保持与真实对象相同的接口 |
| 职责 | 协调多个子系统 | 控制单个对象的访问 |
| 使用场景 | 子系统复杂化简 | 访问控制、延迟加载、缓存 |
6.4 本章总结
6.4.1 核心概念回顾
- 外观模式： 为复杂子系统提供统一的简化接口
- 代理模式： 为对象提供代理以控制对其的访问
- 结构型模式的价值： 通过组合和接口设计解决复杂性问题
6.4.2 最佳实践
外观模式最佳实践：
- 保持外观接口简单直观
- 允许客户端直接访问子系统（可选）
- 避免外观类过于复杂
- 考虑分层外观设计
代理模式最佳实践：
- 保持代理接口与真实对象一致
- 合理选择代理类型
- 注意性能开销
- 考虑代理链的组合使用
6.4.3 实际应用建议
选择合适的模式：
- 需要简化复杂系统时选择外观模式
- 需要控制访问或添加额外功能时选择代理模式
组合使用：
- 可以在外观模式中使用代理模式
- 可以为代理对象创建外观接口
性能考虑：
- 外观模式可能增加方法调用层次
- 代理模式要权衡额外功能与性能开销
6.4.4 注意事项
- 避免过度设计： 不要为了使用模式而使用模式
- 保持接口一致性： 特别是在代理模式中
- 文档化设计决策： 说明为什么选择特定的模式
- 测试覆盖： 确保代理和外观的所有路径都被测试
6.5 练习题
6.5.1 基础练习
外观模式练习：
- 设计一个计算机启动外观，整合CPU、内存、硬盘等子系统
- 实现一个在线购物外观，整合商品、订单、支付、物流等服务
代理模式练习：
- 实现一个文件访问的保护代理，控制不同用户的文件操作权限
- 创建一个数据库连接的虚拟代理，实现连接池管理
6.5.2 进阶练习
组合模式练习：
- 设计一个Web服务外观，内部使用缓存代理、日志代理等
- 实现一个分布式缓存系统，使用代理模式处理网络通信
性能优化练习：
- 比较直接访问与使用代理的性能差异
- 设计一个智能外观，根据负载自动选择最优的子系统
6.5.3 思考题
- 在什么情况下外观模式可能成为反模式？如何避免？
- 代理模式与装饰器模式的区别是什么？何时选择哪个？
- 如何设计一个既是外观又是代理的类？
- 在微服务架构中，外观模式和代理模式分别扮演什么角色？
下一章预告： 接下来我们将学习组合模式和享元模式，探讨如何处理树形结构和优化内存使用。
