Serverless架构模式与最佳实践：构建高效、可扩展的无服务器应用

import json
import boto3
import azure.functions as func
from google.cloud import functions_v1
from dataclasses import dataclass, asdict
from typing import Dict, List, Optional, Any
from datetime import datetime, timedelta
import logging

@dataclass
class FunctionMetrics:
    """函数指标数据"""
    function_name: str
    invocations: int
    duration_avg: float
    duration_p99: float
    error_rate: float
    cold_start_rate: float
    memory_utilization: float
    cost: float
    last_updated: datetime

@dataclass
class ArchitecturePattern:
    """架构模式定义"""
    pattern_name: str
    description: str
    use_cases: List[str]
    components: List[str]
    benefits: List[str]
    challenges: List[str]
    implementation_guide: str

class ServerlessArchitectureAnalyzer:
    """Serverless架构分析器"""
    
    def __init__(self, cloud_provider: str = "aws"):
        self.cloud_provider = cloud_provider
        self.logger = logging.getLogger(__name__)
        self._init_clients()
        
    def _init_clients(self):
        """初始化云服务客户端"""
        if self.cloud_provider == "aws":
            self.lambda_client = boto3.client('lambda')
            self.cloudwatch_client = boto3.client('cloudwatch')
            self.apigateway_client = boto3.client('apigateway')
        elif self.cloud_provider == "azure":
            # Azure Functions 客户端初始化
            pass
        elif self.cloud_provider == "gcp":
            # Google Cloud Functions 客户端初始化
            pass
    
    def analyze_function_performance(self, function_name: str, 
                                   days: int = 7) -> FunctionMetrics:
        """分析函数性能指标"""
        try:
            end_time = datetime.utcnow()
            start_time = end_time - timedelta(days=days)
            
            if self.cloud_provider == "aws":
                return self._analyze_aws_lambda(function_name, start_time, end_time)
            elif self.cloud_provider == "azure":
                return self._analyze_azure_function(function_name, start_time, end_time)
            elif self.cloud_provider == "gcp":
                return self._analyze_gcp_function(function_name, start_time, end_time)
                
        except Exception as e:
            self.logger.error(f"分析函数性能失败: {e}")
            raise
    
    def _analyze_aws_lambda(self, function_name: str, 
                           start_time: datetime, 
                           end_time: datetime) -> FunctionMetrics:
        """分析AWS Lambda函数"""
        # 获取调用次数
        invocations = self._get_cloudwatch_metric(
            'AWS/Lambda', 'Invocations', 
            [{'Name': 'FunctionName', 'Value': function_name}],
            start_time, end_time, 'Sum'
        )
        
        # 获取平均执行时间
        duration_avg = self._get_cloudwatch_metric(
            'AWS/Lambda', 'Duration',
            [{'Name': 'FunctionName', 'Value': function_name}],
            start_time, end_time, 'Average'
        )
        
        # 获取P99执行时间
        duration_p99 = self._get_cloudwatch_metric(
            'AWS/Lambda', 'Duration',
            [{'Name': 'FunctionName', 'Value': function_name}],
            start_time, end_time, 'ExtendedStatistics', ['p99']
        )
        
        # 获取错误率
        errors = self._get_cloudwatch_metric(
            'AWS/Lambda', 'Errors',
            [{'Name': 'FunctionName', 'Value': function_name}],
            start_time, end_time, 'Sum'
        )
        error_rate = (errors / invocations * 100) if invocations > 0 else 0
        
        # 获取冷启动率（通过InitDuration指标估算）
        cold_starts = self._get_cloudwatch_metric(
            'AWS/Lambda', 'InitDuration',
            [{'Name': 'FunctionName', 'Value': function_name}],
            start_time, end_time, 'SampleCount'
        )
        cold_start_rate = (cold_starts / invocations * 100) if invocations > 0 else 0
        
        # 计算成本（简化计算）
        function_config = self.lambda_client.get_function_configuration(
            FunctionName=function_name
        )
        memory_mb = function_config['MemorySize']
        gb_seconds = (duration_avg / 1000) * (memory_mb / 1024) * invocations
        cost = gb_seconds * 0.0000166667  # AWS Lambda定价
        
        return FunctionMetrics(
            function_name=function_name,
            invocations=int(invocations),
            duration_avg=duration_avg,
            duration_p99=duration_p99,
            error_rate=error_rate,
            cold_start_rate=cold_start_rate,
            memory_utilization=0.0,  # 需要额外指标计算
            cost=cost,
            last_updated=datetime.utcnow()
        )
    
    def _get_cloudwatch_metric(self, namespace: str, metric_name: str,
                              dimensions: List[Dict], start_time: datetime,
                              end_time: datetime, statistic: str,
                              extended_statistics: List[str] = None) -> float:
        """获取CloudWatch指标"""
        try:
            params = {
                'Namespace': namespace,
                'MetricName': metric_name,
                'Dimensions': dimensions,
                'StartTime': start_time,
                'EndTime': end_time,
                'Period': 3600,  # 1小时
            }
            
            if extended_statistics:
                params['ExtendedStatistics'] = extended_statistics
            else:
                params['Statistics'] = [statistic]
            
            response = self.cloudwatch_client.get_metric_statistics(**params)
            
            if response['Datapoints']:
                if extended_statistics:
                    return sum(dp['ExtendedStatistics'][extended_statistics[0]] 
                             for dp in response['Datapoints']) / len(response['Datapoints'])
                else:
                    return sum(dp[statistic] for dp in response['Datapoints'])
            return 0.0
            
        except Exception as e:
            self.logger.error(f"获取CloudWatch指标失败: {e}")
            return 0.0
    
    def get_architecture_patterns(self) -> List[ArchitecturePattern]:
        """获取Serverless架构模式"""
        patterns = [
            ArchitecturePattern(
                pattern_name="API后端模式",
                description="使用函数计算构建RESTful API后端服务",
                use_cases=["Web应用后端", "移动应用API", "第三方集成"],
                components=["API Gateway", "Lambda Functions", "数据库"],
                benefits=["自动扩缩容", "按需付费", "高可用性"],
                challenges=["冷启动延迟", "状态管理", "调试复杂"],
                implementation_guide="通过API Gateway触发Lambda函数处理HTTP请求"
            ),
            ArchitecturePattern(
                pattern_name="事件驱动处理",
                description="基于事件触发的异步数据处理",
                use_cases=["文件处理", "数据ETL", "消息处理"],
                components=["事件源", "Lambda Functions", "存储服务"],
                benefits=["解耦架构", "弹性扩展", "容错能力"],
                challenges=["事件顺序", "重复处理", "错误处理"],
                implementation_guide="配置事件源触发器，实现异步处理逻辑"
            ),
            ArchitecturePattern(
                pattern_name="CQRS模式",
                description="命令查询责任分离的Serverless实现",
                use_cases=["复杂业务逻辑", "读写分离", "事件溯源"],
                components=["命令函数", "查询函数", "事件存储"],
                benefits=["职责分离", "性能优化", "可扩展性"],
                challenges=["数据一致性", "复杂性增加", "事件管理"],
                implementation_guide="分别实现命令和查询处理函数"
            ),
            ArchitecturePattern(
                pattern_name="微服务编排",
                description="使用工作流编排多个微服务",
                use_cases=["业务流程", "数据管道", "批处理任务"],
                components=["Step Functions", "Lambda Functions", "状态机"],
                benefits=["可视化流程", "错误处理", "状态管理"],
                challenges=["复杂度管理", "调试困难", "成本控制"],
                implementation_guide="定义状态机编排多个函数执行"
            )
        ]
        return patterns
    
    def optimize_function_configuration(self, function_name: str,
                                      metrics: FunctionMetrics) -> Dict[str, Any]:
        """优化函数配置建议"""
        recommendations = {
            "memory_optimization": {},
            "timeout_optimization": {},
            "concurrency_optimization": {},
            "cost_optimization": {}
        }
        
        # 内存优化建议
        if metrics.memory_utilization < 50:
            recommendations["memory_optimization"] = {
                "action": "减少内存分配",
                "current_memory": "获取当前配置",
                "recommended_memory": "建议减少20-30%",
                "expected_savings": "预计节省成本"
            }
        elif metrics.memory_utilization > 80:
            recommendations["memory_optimization"] = {
                "action": "增加内存分配",
                "reason": "内存使用率过高，可能影响性能",
                "recommended_increase": "建议增加20-50%"
            }
        
        # 超时优化建议
        if metrics.duration_p99 > 25000:  # 25秒
            recommendations["timeout_optimization"] = {
                "action": "检查函数逻辑",
                "reason": "执行时间过长，可能需要优化",
                "suggestions": ["代码优化", "异步处理", "拆分函数"]
            }
        
        # 并发优化建议
        if metrics.error_rate > 5:
            recommendations["concurrency_optimization"] = {
                "action": "设置预留并发",
                "reason": "错误率较高，可能存在并发限制",
                "recommended_concurrency": "基于峰值流量设置"
            }
        
        # 成本优化建议
        if metrics.cost > 100:  # 月成本超过100美元
            recommendations["cost_optimization"] = {
                "action": "成本优化分析",
                "suggestions": [
                    "优化函数执行时间",
                    "调整内存配置",
                    "使用预留容量",
                    "考虑替代方案"
                ]
            }
        
        return recommendations
    
    def generate_architecture_report(self, functions: List[str]) -> Dict[str, Any]:
        """生成架构分析报告"""
        report = {
            "summary": {
                "total_functions": len(functions),
                "analysis_date": datetime.utcnow().isoformat(),
                "cloud_provider": self.cloud_provider
            },
            "function_metrics": [],
            "optimization_recommendations": [],
            "architecture_patterns": [],
            "best_practices": []
        }
        
        # 分析每个函数
        for function_name in functions:
            try:
                metrics = self.analyze_function_performance(function_name)
                report["function_metrics"].append(asdict(metrics))
                
                # 生成优化建议
                recommendations = self.optimize_function_configuration(
                    function_name, metrics
                )
                report["optimization_recommendations"].append({
                    "function_name": function_name,
                    "recommendations": recommendations
                })
                
            except Exception as e:
                self.logger.error(f"分析函数 {function_name} 失败: {e}")
        
        # 添加架构模式
        patterns = self.get_architecture_patterns()
        report["architecture_patterns"] = [asdict(pattern) for pattern in patterns]
        
        # 添加最佳实践
        report["best_practices"] = self._get_best_practices()
        
        return report
    
    def _get_best_practices(self) -> List[Dict[str, str]]:
        """获取最佳实践建议"""
        return [
            {
                "category": "函数设计",
                "practice": "保持函数单一职责",
                "description": "每个函数只处理一个特定的业务逻辑"
            },
            {
                "category": "性能优化",
                "practice": "优化冷启动时间",
                "description": "减少依赖包大小，使用连接池，预热函数"
            },
            {
                "category": "错误处理",
                "practice": "实现重试机制",
                "description": "配置死信队列，实现指数退避重试"
            },
            {
                "category": "监控运维",
                "practice": "完善监控告警",
                "description": "监控关键指标，设置合理的告警阈值"
            },
            {
                "category": "安全管理",
                "practice": "最小权限原则",
                "description": "为函数分配最小必要的权限"
            }
        ]

def main():
    """主函数示例"""
    # 初始化分析器
    analyzer = ServerlessArchitectureAnalyzer("aws")
    
    # 分析函数列表
    functions = ["user-service", "order-service", "payment-service"]
    
    # 生成架构报告
    report = analyzer.generate_architecture_report(functions)
    
    # 输出报告
    print("=== Serverless架构分析报告 ===")
    print(f"分析函数数量: {report['summary']['total_functions']}")
    print(f"分析时间: {report['summary']['analysis_date']}")
    
    # 输出函数指标
    for metrics in report["function_metrics"]:
        print(f"\n函数: {metrics['function_name']}")
        print(f"  调用次数: {metrics['invocations']}")
        print(f"  平均执行时间: {metrics['duration_avg']:.2f}ms")
        print(f"  错误率: {metrics['error_rate']:.2f}%")
        print(f"  冷启动率: {metrics['cold_start_rate']:.2f}%")
        print(f"  成本: ${metrics['cost']:.2f}")

if __name__ == "__main__":
    main()

from typing import Callable, Any, Dict, List
import asyncio
import json
from functools import wraps

class FunctionComposer:
    """函数组合器"""
    
    def __init__(self):
        self.functions = {}
        self.middleware = []
    
    def register(self, name: str):
        """注册函数装饰器"""
        def decorator(func: Callable):
            self.functions[name] = func
            return func
        return decorator
    
    def middleware(self, middleware_func: Callable):
        """添加中间件"""
        self.middleware.append(middleware_func)
        return middleware_func
    
    def compose(self, *function_names: str) -> Callable:
        """组合多个函数"""
        def composed_function(event: Dict[str, Any]) -> Dict[str, Any]:
            result = event
            
            # 应用中间件
            for middleware in self.middleware:
                result = middleware(result)
            
            # 依次执行函数
            for name in function_names:
                if name in self.functions:
                    result = self.functions[name](result)
                else:
                    raise ValueError(f"函数 {name} 未注册")
            
            return result
        
        return composed_function
    
    async def compose_async(self, *function_names: str) -> Callable:
        """异步函数组合"""
        async def composed_function(event: Dict[str, Any]) -> Dict[str, Any]:
            result = event
            
            # 应用中间件
            for middleware in self.middleware:
                if asyncio.iscoroutinefunction(middleware):
                    result = await middleware(result)
                else:
                    result = middleware(result)
            
            # 依次执行函数
            for name in function_names:
                if name in self.functions:
                    func = self.functions[name]
                    if asyncio.iscoroutinefunction(func):
                        result = await func(result)
                    else:
                        result = func(result)
                else:
                    raise ValueError(f"函数 {name} 未注册")
            
            return result
        
        return composed_function

# 使用示例
composer = FunctionComposer()

@composer.register("validate")
def validate_input(event: Dict[str, Any]) -> Dict[str, Any]:
    """输入验证"""
    if "user_id" not in event:
        raise ValueError("缺少user_id参数")
    return event

@composer.register("process")
def process_data(event: Dict[str, Any]) -> Dict[str, Any]:
    """数据处理"""
    event["processed"] = True
    event["timestamp"] = "2024-06-15T10:00:00Z"
    return event

@composer.register("save")
def save_result(event: Dict[str, Any]) -> Dict[str, Any]:
    """保存结果"""
    event["saved"] = True
    return event

# 组合函数
pipeline = composer.compose("validate", "process", "save")

import re
from typing import Dict, Any, Callable, List, Optional
from dataclasses import dataclass
from enum import Enum

class EventType(Enum):
    """事件类型"""
    HTTP_REQUEST = "http_request"
    S3_OBJECT_CREATED = "s3_object_created"
    DYNAMODB_STREAM = "dynamodb_stream"
    SQS_MESSAGE = "sqs_message"
    SCHEDULED = "scheduled"
    CUSTOM = "custom"

@dataclass
class EventRoute:
    """事件路由配置"""
    pattern: str
    handler: Callable
    event_type: EventType
    priority: int = 0
    middleware: List[Callable] = None

class EventRouter:
    """事件路由器"""
    
    def __init__(self):
        self.routes: List[EventRoute] = []
        self.global_middleware: List[Callable] = []
    
    def add_route(self, pattern: str, event_type: EventType, 
                  priority: int = 0, middleware: List[Callable] = None):
        """添加路由装饰器"""
        def decorator(handler: Callable):
            route = EventRoute(
                pattern=pattern,
                handler=handler,
                event_type=event_type,
                priority=priority,
                middleware=middleware or []
            )
            self.routes.append(route)
            # 按优先级排序
            self.routes.sort(key=lambda r: r.priority, reverse=True)
            return handler
        return decorator
    
    def add_middleware(self, middleware: Callable):
        """添加全局中间件"""
        self.global_middleware.append(middleware)
        return middleware
    
    def route_event(self, event: Dict[str, Any], context: Any = None) -> Dict[str, Any]:
        """路由事件到对应处理器"""
        event_type = self._detect_event_type(event)
        event_path = self._extract_event_path(event, event_type)
        
        # 查找匹配的路由
        for route in self.routes:
            if route.event_type == event_type and self._match_pattern(route.pattern, event_path):
                return self._execute_handler(route, event, context)
        
        # 没有找到匹配的路由
        return {
            "statusCode": 404,
            "body": json.dumps({"error": "No matching route found"})
        }
    
    def _detect_event_type(self, event: Dict[str, Any]) -> EventType:
        """检测事件类型"""
        if "httpMethod" in event or "requestContext" in event:
            return EventType.HTTP_REQUEST
        elif "Records" in event:
            records = event["Records"]
            if records and "s3" in records[0]:
                return EventType.S3_OBJECT_CREATED
            elif records and "dynamodb" in records[0]:
                return EventType.DYNAMODB_STREAM
            elif records and "eventSource" in records[0] and records[0]["eventSource"] == "aws:sqs":
                return EventType.SQS_MESSAGE
        elif "source" in event and event["source"] == "aws.events":
            return EventType.SCHEDULED
        else:
            return EventType.CUSTOM
    
    def _extract_event_path(self, event: Dict[str, Any], event_type: EventType) -> str:
        """提取事件路径"""
        if event_type == EventType.HTTP_REQUEST:
            return event.get("path", "/")
        elif event_type == EventType.S3_OBJECT_CREATED:
            return event["Records"][0]["s3"]["object"]["key"]
        elif event_type == EventType.SQS_MESSAGE:
            return event["Records"][0].get("messageAttributes", {}).get("route", {}).get("stringValue", "default")
        else:
            return event.get("detail-type", "default")
    
    def _match_pattern(self, pattern: str, path: str) -> bool:
        """匹配路由模式"""
        # 支持通配符和正则表达式
        if "*" in pattern:
            regex_pattern = pattern.replace("*", ".*")
            return re.match(f"^{regex_pattern}$", path) is not None
        else:
            return re.match(f"^{pattern}$", path) is not None
    
    def _execute_handler(self, route: EventRoute, event: Dict[str, Any], context: Any) -> Dict[str, Any]:
        """执行处理器"""
        try:
            # 应用全局中间件
            for middleware in self.global_middleware:
                event = middleware(event, context)
            
            # 应用路由中间件
            for middleware in route.middleware:
                event = middleware(event, context)
            
            # 执行处理器
            result = route.handler(event, context)
            
            # 确保返回格式正确
            if isinstance(result, dict) and "statusCode" in result:
                return result
            else:
                return {
                    "statusCode": 200,
                    "body": json.dumps(result) if not isinstance(result, str) else result
                }
                
        except Exception as e:
            return {
                "statusCode": 500,
                "body": json.dumps({"error": str(e)})
            }

# 使用示例
router = EventRouter()

@router.add_middleware
def cors_middleware(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """CORS中间件"""
    event["_cors_enabled"] = True
    return event

@router.add_middleware
def auth_middleware(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """认证中间件"""
    # 简化的认证逻辑
    headers = event.get("headers", {})
    if "authorization" not in headers:
        raise Exception("Missing authorization header")
    return event

@router.add_route("/api/users/*", EventType.HTTP_REQUEST, priority=10)
def handle_user_api(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """处理用户API请求"""
    method = event.get("httpMethod", "GET")
    path = event.get("path", "")
    
    if method == "GET":
        return {"message": f"Getting user from {path}"}
    elif method == "POST":
        return {"message": f"Creating user at {path}"}
    else:
        return {"error": "Method not allowed"}

@router.add_route("*.jpg", EventType.S3_OBJECT_CREATED)
def handle_image_upload(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """处理图片上传"""
    bucket = event["Records"][0]["s3"]["bucket"]["name"]
    key = event["Records"][0]["s3"]["object"]["key"]
    
    return {
        "message": f"Processing image {key} from bucket {bucket}",
        "action": "thumbnail_generation"
    }

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """Lambda入口函数"""
    return router.route_event(event, context)

import time
import psutil
import threading
from typing import Dict, Any, Callable, Optional
from functools import wraps
from dataclasses import dataclass
import json

@dataclass
class PerformanceMetrics:
    """性能指标"""
    execution_time: float
    memory_usage: float
    cpu_usage: float
    cold_start: bool
    error_occurred: bool
    error_message: Optional[str] = None

class FunctionOptimizer:
    """函数性能优化器"""
    
    def __init__(self):
        self.metrics_history: Dict[str, List[PerformanceMetrics]] = {}
        self.connection_pool = {}
        self.cache = {}
        self.is_cold_start = True
    
    def performance_monitor(self, function_name: str):
        """性能监控装饰器"""
        def decorator(func: Callable):
            @wraps(func)
            def wrapper(*args, **kwargs):
                start_time = time.time()
                start_memory = psutil.Process().memory_info().rss / 1024 / 1024  # MB
                start_cpu = psutil.cpu_percent()
                
                error_occurred = False
                error_message = None
                result = None
                
                try:
                    result = func(*args, **kwargs)
                except Exception as e:
                    error_occurred = True
                    error_message = str(e)
                    raise
                finally:
                    end_time = time.time()
                    end_memory = psutil.Process().memory_info().rss / 1024 / 1024
                    end_cpu = psutil.cpu_percent()
                    
                    metrics = PerformanceMetrics(
                        execution_time=(end_time - start_time) * 1000,  # ms
                        memory_usage=end_memory - start_memory,
                        cpu_usage=end_cpu - start_cpu,
                        cold_start=self.is_cold_start,
                        error_occurred=error_occurred,
                        error_message=error_message
                    )
                    
                    self._record_metrics(function_name, metrics)
                    self.is_cold_start = False
                
                return result
            return wrapper
        return decorator
    
    def _record_metrics(self, function_name: str, metrics: PerformanceMetrics):
        """记录性能指标"""
        if function_name not in self.metrics_history:
            self.metrics_history[function_name] = []
        
        self.metrics_history[function_name].append(metrics)
        
        # 保持最近100次记录
        if len(self.metrics_history[function_name]) > 100:
            self.metrics_history[function_name] = self.metrics_history[function_name][-100:]
    
    def connection_pool_manager(self, pool_name: str, create_connection: Callable):
        """连接池管理装饰器"""
        def decorator(func: Callable):
            @wraps(func)
            def wrapper(*args, **kwargs):
                if pool_name not in self.connection_pool:
                    self.connection_pool[pool_name] = create_connection()
                
                # 将连接注入到函数参数中
                kwargs['connection'] = self.connection_pool[pool_name]
                return func(*args, **kwargs)
            return wrapper
        return decorator
    
    def cache_result(self, ttl: int = 300, key_func: Callable = None):
        """结果缓存装饰器"""
        def decorator(func: Callable):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # 生成缓存键
                if key_func:
                    cache_key = key_func(*args, **kwargs)
                else:
                    cache_key = f"{func.__name__}:{hash(str(args) + str(kwargs))}"
                
                # 检查缓存
                if cache_key in self.cache:
                    cached_result, timestamp = self.cache[cache_key]
                    if time.time() - timestamp < ttl:
                        return cached_result
                
                # 执行函数并缓存结果
                result = func(*args, **kwargs)
                self.cache[cache_key] = (result, time.time())
                
                return result
            return wrapper
        return decorator
    
    def async_processor(self, max_workers: int = 5):
        """异步处理装饰器"""
        def decorator(func: Callable):
            @wraps(func)
            def wrapper(*args, **kwargs):
                import concurrent.futures
                
                # 检查是否有批量数据需要处理
                if 'batch_data' in kwargs:
                    batch_data = kwargs.pop('batch_data')
                    
                    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
                        futures = []
                        for item in batch_data:
                            future = executor.submit(func, item, *args, **kwargs)
                            futures.append(future)
                        
                        results = []
                        for future in concurrent.futures.as_completed(futures):
                            try:
                                result = future.result()
                                results.append(result)
                            except Exception as e:
                                results.append({"error": str(e)})
                        
                        return results
                else:
                    return func(*args, **kwargs)
            return wrapper
        return decorator
    
    def get_optimization_recommendations(self, function_name: str) -> Dict[str, Any]:
        """获取优化建议"""
        if function_name not in self.metrics_history:
            return {"error": "No metrics available for this function"}
        
        metrics = self.metrics_history[function_name]
        if not metrics:
            return {"error": "No metrics data"}
        
        # 计算统计信息
        avg_execution_time = sum(m.execution_time for m in metrics) / len(metrics)
        avg_memory_usage = sum(m.memory_usage for m in metrics) / len(metrics)
        cold_start_rate = sum(1 for m in metrics if m.cold_start) / len(metrics) * 100
        error_rate = sum(1 for m in metrics if m.error_occurred) / len(metrics) * 100
        
        recommendations = {
            "performance_summary": {
                "avg_execution_time_ms": round(avg_execution_time, 2),
                "avg_memory_usage_mb": round(avg_memory_usage, 2),
                "cold_start_rate_percent": round(cold_start_rate, 2),
                "error_rate_percent": round(error_rate, 2)
            },
            "recommendations": []
        }
        
        # 生成优化建议
        if avg_execution_time > 5000:  # 5秒
            recommendations["recommendations"].append({
                "type": "performance",
                "issue": "执行时间过长",
                "suggestion": "考虑优化算法、使用缓存或异步处理",
                "priority": "high"
            })
        
        if avg_memory_usage > 100:  # 100MB
            recommendations["recommendations"].append({
                "type": "memory",
                "issue": "内存使用量较高",
                "suggestion": "优化数据结构、及时释放资源",
                "priority": "medium"
            })
        
        if cold_start_rate > 20:  # 20%
            recommendations["recommendations"].append({
                "type": "cold_start",
                "issue": "冷启动率较高",
                "suggestion": "使用预留并发、减少依赖包大小、实现预热机制",
                "priority": "high"
            })
        
        if error_rate > 5:  # 5%
            recommendations["recommendations"].append({
                "type": "reliability",
                "issue": "错误率较高",
                "suggestion": "加强错误处理、实现重试机制、完善监控",
                "priority": "critical"
            })
        
        return recommendations

# 使用示例
optimizer = FunctionOptimizer()

# 数据库连接创建函数
def create_db_connection():
    """创建数据库连接（示例）"""
    return {"host": "localhost", "port": 5432, "connected": True}

@optimizer.performance_monitor("user_service")
@optimizer.connection_pool_manager("db_pool", create_db_connection)
@optimizer.cache_result(ttl=300)
def get_user_profile(user_id: str, connection=None) -> Dict[str, Any]:
    """获取用户资料"""
    # 模拟数据库查询
    time.sleep(0.1)  # 模拟查询延迟
    
    return {
        "user_id": user_id,
        "name": f"User {user_id}",
        "email": f"user{user_id}@example.com",
        "connection_info": connection
    }

@optimizer.performance_monitor("batch_processor")
@optimizer.async_processor(max_workers=10)
def process_user_data(user_data: Dict[str, Any]) -> Dict[str, Any]:
    """处理用户数据"""
    # 模拟数据处理
    time.sleep(0.05)
    
    return {
        "processed": True,
        "user_id": user_data.get("user_id"),
        "timestamp": time.time()
    }

import json
import uuid
import asyncio
from typing import Dict, Any, List, Callable, Optional, Union
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from enum import Enum
import boto3
from abc import ABC, abstractmethod

class EventStatus(Enum):
    """事件状态"""
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"
    RETRYING = "retrying"

@dataclass
class Event:
    """事件数据结构"""
    id: str
    type: str
    source: str
    data: Dict[str, Any]
    timestamp: datetime
    correlation_id: Optional[str] = None
    status: EventStatus = EventStatus.PENDING
    retry_count: int = 0
    max_retries: int = 3
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return {
            **asdict(self),
            'timestamp': self.timestamp.isoformat(),
            'status': self.status.value
        }

class EventHandler(ABC):
    """事件处理器抽象基类"""
    
    @abstractmethod
    async def handle(self, event: Event) -> Dict[str, Any]:
        """处理事件"""
        pass
    
    @abstractmethod
    def can_handle(self, event: Event) -> bool:
        """判断是否能处理该事件"""
        pass

class EventBus:
    """事件总线"""
    
    def __init__(self, dead_letter_queue: Optional[str] = None):
        self.handlers: List[EventHandler] = []
        self.middleware: List[Callable] = []
        self.dead_letter_queue = dead_letter_queue
        self.event_store: List[Event] = []
        
        # AWS服务客户端
        self.sqs = boto3.client('sqs')
        self.sns = boto3.client('sns')
    
    def register_handler(self, handler: EventHandler):
        """注册事件处理器"""
        self.handlers.append(handler)
    
    def add_middleware(self, middleware: Callable):
        """添加中间件"""
        self.middleware.append(middleware)
    
    async def publish(self, event: Event):
        """发布事件"""
        # 应用中间件
        for middleware in self.middleware:
            event = await middleware(event) if asyncio.iscoroutinefunction(middleware) else middleware(event)
        
        # 存储事件
        self.event_store.append(event)
        
        # 处理事件
        await self._process_event(event)
    
    async def _process_event(self, event: Event):
        """处理事件"""
        event.status = EventStatus.PROCESSING
        
        # 查找合适的处理器
        suitable_handlers = [h for h in self.handlers if h.can_handle(event)]
        
        if not suitable_handlers:
            await self._send_to_dead_letter_queue(event, "No suitable handler found")
            return
        
        # 并行处理（如果有多个处理器）
        tasks = []
        for handler in suitable_handlers:
            task = asyncio.create_task(self._handle_with_retry(handler, event))
            tasks.append(task)
        
        try:
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # 检查处理结果
            success_count = sum(1 for r in results if not isinstance(r, Exception))
            if success_count > 0:
                event.status = EventStatus.COMPLETED
            else:
                event.status = EventStatus.FAILED
                await self._send_to_dead_letter_queue(event, f"All handlers failed: {results}")
                
        except Exception as e:
            event.status = EventStatus.FAILED
            await self._send_to_dead_letter_queue(event, str(e))
    
    async def _handle_with_retry(self, handler: EventHandler, event: Event) -> Dict[str, Any]:
        """带重试的事件处理"""
        last_exception = None
        
        for attempt in range(event.max_retries + 1):
            try:
                event.retry_count = attempt
                if attempt > 0:
                    event.status = EventStatus.RETRYING
                    # 指数退避
                    await asyncio.sleep(2 ** attempt)
                
                result = await handler.handle(event)
                return result
                
            except Exception as e:
                last_exception = e
                if attempt == event.max_retries:
                    raise e
        
        raise last_exception
    
    async def _send_to_dead_letter_queue(self, event: Event, error_message: str):
        """发送到死信队列"""
        if self.dead_letter_queue:
            try:
                message = {
                    "event": event.to_dict(),
                    "error": error_message,
                    "failed_at": datetime.now(timezone.utc).isoformat()
                }
                
                self.sqs.send_message(
                    QueueUrl=self.dead_letter_queue,
                    MessageBody=json.dumps(message)
                )
            except Exception as e:
                print(f"Failed to send to dead letter queue: {e}")

# 具体事件处理器实现
class UserRegistrationHandler(EventHandler):
    """用户注册事件处理器"""
    
    async def handle(self, event: Event) -> Dict[str, Any]:
        """处理用户注册事件"""
        user_data = event.data
        
        # 发送欢迎邮件
        await self._send_welcome_email(user_data)
        
        # 创建用户配置文件
        await self._create_user_profile(user_data)
        
        # 分配默认权限
        await self._assign_default_permissions(user_data)
        
        return {"status": "success", "user_id": user_data["user_id"]}
    
    def can_handle(self, event: Event) -> bool:
        """判断是否能处理该事件"""
        return event.type == "user.registered"
    
    async def _send_welcome_email(self, user_data: Dict[str, Any]):
        """发送欢迎邮件"""
        # 模拟邮件发送
        await asyncio.sleep(0.1)
        print(f"Welcome email sent to {user_data['email']}")
    
    async def _create_user_profile(self, user_data: Dict[str, Any]):
        """创建用户配置文件"""
        # 模拟数据库操作
        await asyncio.sleep(0.05)
        print(f"User profile created for {user_data['user_id']}")
    
    async def _assign_default_permissions(self, user_data: Dict[str, Any]):
        """分配默认权限"""
        # 模拟权限分配
        await asyncio.sleep(0.02)
        print(f"Default permissions assigned to {user_data['user_id']}")

class OrderProcessingHandler(EventHandler):
    """订单处理事件处理器"""
    
    async def handle(self, event: Event) -> Dict[str, Any]:
        """处理订单事件"""
        order_data = event.data
        
        if event.type == "order.created":
            return await self._process_new_order(order_data)
        elif event.type == "order.paid":
            return await self._process_payment(order_data)
        elif event.type == "order.shipped":
            return await self._process_shipment(order_data)
        
        return {"status": "unknown_event_type"}
    
    def can_handle(self, event: Event) -> bool:
        """判断是否能处理该事件"""
        return event.type.startswith("order.")
    
    async def _process_new_order(self, order_data: Dict[str, Any]) -> Dict[str, Any]:
        """处理新订单"""
        # 库存检查
        await self._check_inventory(order_data)
        
        # 计算价格
        total_price = await self._calculate_price(order_data)
        
        # 创建支付请求
        payment_id = await self._create_payment_request(order_data, total_price)
        
        return {
            "status": "order_processed",
            "order_id": order_data["order_id"],
            "payment_id": payment_id,
            "total_price": total_price
        }
    
    async def _check_inventory(self, order_data: Dict[str, Any]):
        """检查库存"""
        await asyncio.sleep(0.1)
        print(f"Inventory checked for order {order_data['order_id']}")
    
    async def _calculate_price(self, order_data: Dict[str, Any]) -> float:
        """计算价格"""
        await asyncio.sleep(0.05)
        return sum(item["price"] * item["quantity"] for item in order_data["items"])
    
    async def _create_payment_request(self, order_data: Dict[str, Any], total_price: float) -> str:
        """创建支付请求"""
        await asyncio.sleep(0.1)
        payment_id = str(uuid.uuid4())
        print(f"Payment request created: {payment_id} for ${total_price}")
        return payment_id
    
    async def _process_payment(self, order_data: Dict[str, Any]) -> Dict[str, Any]:
        """处理支付"""
        await asyncio.sleep(0.2)
        print(f"Payment processed for order {order_data['order_id']}")
        return {"status": "payment_processed"}
    
    async def _process_shipment(self, order_data: Dict[str, Any]) -> Dict[str, Any]:
        """处理发货"""
        await asyncio.sleep(0.15)
        print(f"Shipment processed for order {order_data['order_id']}")
        return {"status": "shipment_processed"}

# 中间件示例
async def logging_middleware(event: Event) -> Event:
    """日志中间件"""
    print(f"Processing event: {event.type} - {event.id}")
    return event

async def validation_middleware(event: Event) -> Event:
    """验证中间件"""
    if not event.data:
        raise ValueError("Event data cannot be empty")
    return event

# 使用示例
async def main():
    """主函数示例"""
    # 创建事件总线
    event_bus = EventBus(dead_letter_queue="https://sqs.region.amazonaws.com/account/dlq")
    
    # 注册处理器
    event_bus.register_handler(UserRegistrationHandler())
    event_bus.register_handler(OrderProcessingHandler())
    
    # 添加中间件
    event_bus.add_middleware(logging_middleware)
    event_bus.add_middleware(validation_middleware)
    
    # 创建并发布事件
    user_event = Event(
        id=str(uuid.uuid4()),
        type="user.registered",
        source="user-service",
        data={
            "user_id": "user123",
            "email": "user@example.com",
            "name": "John Doe"
        },
        timestamp=datetime.now(timezone.utc),
        correlation_id=str(uuid.uuid4())
    )
    
    order_event = Event(
        id=str(uuid.uuid4()),
        type="order.created",
        source="order-service",
        data={
            "order_id": "order456",
            "user_id": "user123",
            "items": [
                {"product_id": "prod1", "quantity": 2, "price": 29.99},
                {"product_id": "prod2", "quantity": 1, "price": 49.99}
            ]
        },
        timestamp=datetime.now(timezone.utc),
        correlation_id=str(uuid.uuid4())
    )
    
    # 发布事件
    await event_bus.publish(user_event)
    await event_bus.publish(order_event)
    
    print("Events published and processed")

if __name__ == "__main__":
    asyncio.run(main())