《精通Python设计模式》精读:微服务架构模式

import asyncioimport jsonfrom abc import ABC, abstractmethodfrom typing import Dict, Any, Callablefrom dataclasses import dataclass@dataclassclass ServiceMessage:    """服务消息"""    service_name: str    operation: str    data: Dict[str, Any]    correlation_id: str = Noneclass Service(ABC):    """微服务基类"""    def __init__(self, name: str):        self.name = name        self.handlers: Dict[str, Callable] = {}    def register_handler(self, operation: str, handler: Callable):        """注册操作处理器"""        self.handlers[operation] = handler    @abstractmethod    async def process_message(self, message: ServiceMessage) -> Any:        passclass UserService(Service):    """用户服务"""    def __init__(self):        super().__init__("user-service")        self.users = {            "1": {"id": "1", "name": "Alice", "email": "alice@example.com"},            "2": {"id": "2", "name": "Bob", "email": "bob@example.com"}        }    async def process_message(self, message: ServiceMessage) -> Any:        if message.operation in self.handlers:            return await self.handlers[message.operation](message.data)        else:            raise ValueError(f"Unknown operation: {message.operation}")    async def get_user(self, data: Dict[str, Any]):        user_id = data.get("user_id")        return self.users.get(user_id)    async def create_user(self, data: Dict[str, Any]):        user_id = str(len(self.users) + 1)        user = {            "id": user_id,            "name": data.get("name"),            "email": data.get("email")        }        self.users[user_id] = user        return userclass OrderService(Service):    """订单服务"""    def __init__(self):        super().__init__("order-service")        self.orders = []    async def process_message(self, message: ServiceMessage) -> Any:        if message.operation in self.handlers:            return await self.handlers[message.operation](message.data)        else:            raise ValueError(f"Unknown operation: {message.operation}")    async def create_order(self, data: Dict[str, Any]):        order = {            "id": len(self.orders) + 1,            "user_id": data.get("user_id"),            "product": data.get("product"),            "quantity": data.get("quantity", 1)        }        self.orders.append(order)        return orderclass ServiceRegistry:    """服务注册表"""    def __init__(self):        self.services: Dict[str, Service] = {}    def register_service(self, service: Service):        self.services[service.name] = service    def get_service(self, service_name: str) -> Service:        return self.services.get(service_name)class ServiceMesh:    """服务网格/API网关"""    def __init__(self, registry: ServiceRegistry):        self.registry = registry        self.message_queue = asyncio.Queue()    async def send_message(self, message: ServiceMessage) -> Any:        """发送消息到指定服务"""        service = self.registry.get_service(message.service_name)        if service:            return await service.process_message(message)        else:            raise ValueError(f"Service not found: {message.service_name}")# 配置服务并运行async def run_microservices_demo():    # 创建服务注册表    registry = ServiceRegistry()    # 创建并注册用户服务    user_service = UserService()    user_service.register_handler("get_user", user_service.get_user)    user_service.register_handler("create_user", user_service.create_user)    registry.register_service(user_service)    # 创建并注册订单服务    order_service = OrderService()    order_service.register_handler("create_order", order_service.create_order)    registry.register_service(order_service)    # 创建服务网格    mesh = ServiceMesh(registry)    # 演示服务间通信    print("=== 微服务通信演示 ===")    # 创建用户    user_message = ServiceMessage(        service_name="user-service",        operation="create_user",        data={"name": "Charlie", "email": "charlie@example.com"}    )    new_user = await mesh.send_message(user_message)    print(f"创建用户: {new_user}")    # 创建订单    order_message = ServiceMessage(        service_name="order-service",        operation="create_order",        data={"user_id": new_user["id"], "product": "Python书", "quantity": 2}    )    new_order = await mesh.send_message(order_message)    print(f"创建订单: {new_order}")# 运行演示# asyncio.run(run_microservices_demo())

from urllib.parse import urlparseimport timeimport functoolsclass RateLimiter:    """限流器"""    def __init__(self, max_requests: int, time_window: int):        self.max_requests = max_requests        self.time_window = time_window        self.requests = {}    def __call__(self, func):        @functools.wraps(func)        async def wrapper(*args, **kwargs):            client_id = kwargs.get('client_id', 'default')            now = time.time()            if client_id not in self.requests:                self.requests[client_id] = []            # 清理过期请求            self.requests[client_id] = [                req_time for req_time in self.requests[client_id]                if now - req_time < self.time_window            ]            if len(self.requests[client_id]) >= self.max_requests:                raise Exception("请求过于频繁")            self.requests[client_id].append(now)            return await func(*args, **kwargs)        return wrapperclass APIGateway:    """API网关"""    def __init__(self):        self.service_routes = {}        self.rate_limiter = RateLimiter(max_requests=10, time_window=60)  # 1分钟内最多10次请求    def route(self, path: str, service_name: str, service_path: str):        """配置路由"""        self.service_routes[path] = {            'service_name': service_name,            'service_path': service_path        }    @RateLimiter(max_requests=10, time_window=60)    async def handle_request(self, path: str, method: str, data: Dict[str, Any] = None, client_id: str = None):        """处理请求"""        if path in self.service_routes:            route = self.service_routes[path]            # 这里会转发请求到相应的微服务            print(f"转发请求到服务: {route['service_name']}, 路径: {route['service_path']}")            print(f"方法: {method}, 数据: {data}")            # 实际实现中会通过HTTP或消息队列转发请求            return {"status": "forwarded", "service": route['service_name']}        else:            raise ValueError(f"路由未找到: {path}")# 使用API网关gateway = APIGateway()gateway.route("/api/users", "user-service", "/users")gateway.route("/api/orders", "order-service", "/orders")# 模拟API请求# result = await gateway.handle_request("/api/users", "GET", client_id="client_123")

class EventDrivenArchitectureExample:    """事件驱动架构示例，组合多种模式"""    def __init__(self):        # 观察者模式：事件系统        self.event_handlers = {}        # 策略模式：不同的处理策略        self.processing_strategies = {            'validate': self.validate_data,            'transform': self.transform_data,            'save': self.save_data        }        # 命令模式：可撤销的操作        self.command_history = []        # 工厂模式：创建处理器        self.handler_factory = self._create_handler    def _create_handler(self, handler_type: str):        """处理器工厂"""        handlers = {            'email': EmailNotificationHandler,            'sms': SMSNotificationHandler,            'push': PushNotificationHandler        }        return handlers.get(handler_type, DefaultNotificationHandler)()    def validate_data(self, data):        """验证数据"""        print(f"验证数据: {data}")        return True    def transform_data(self, data):        """转换数据"""        print(f"转换数据: {data}")        return data    def save_data(self, data):        """保存数据"""        print(f"保存数据: {data}")        return Trueclass NotificationHandler(ABC):    """通知处理器抽象类"""    @abstractmethod    def send(self, message: str):        passclass EmailNotificationHandler(NotificationHandler):    def send(self, message: str):        print(f"发送邮件通知: {message}")class SMSNotificationHandler(NotificationHandler):    def send(self, message: str):        print(f"发送短信通知: {message}")class PushNotificationHandler(NotificationHandler):    def send(self, message: str):        print(f"发送推送通知: {message}")class DefaultNotificationHandler(NotificationHandler):    def send(self, message: str):        print(f"发送默认通知: {message}")