| 5. 执行失效 | 调用API | 完成失效 | 返回失效ID |

#### 失效路径优化示例

| 10000个 | 50个 | 99.5% | $49.75 | 根目录通配符 |

三、源站优化配置

3.1 多源站负载均衡

#### 源站组故障转移配置

| 健康检查 | 自动 | CloudFront自动检测 | 无需手动配置 |

#### 源站连接优化参数

| SSL协议 | TLSv1.2, TLSv1.3 | TLSv1.2 | 提高安全性 |

#### 自定义头部配置

| X-Cache-Control | 缓存控制 | max-age=3600 | 灵活缓存 |

#### Origin Shield 配置指南

| 自动选择 | 全球业务 | ~5% | 智能路由 |

3.2 S3 源站优化

#### S3 源站优化配置

| 区域选择 | 同区域传输 | 减少延迟 | 免传输费 |

#### S3 存储桶策略配置

| Condition | IP限制 | 可选IP白名单 | 增强 |

#### 智能分层配置建议

| 深度归档 | >180天 | 备份数据 | 节眉95% |

#### S3 与 CloudFront 集成最佳实践

| 生命周期 | 设置过期规则 | 自动清理 | 减少存储成本 |

四、请求优化技术

4.1 请求合并与批处理

#### 请求优化策略

| 延迟加载 | Lazy Loading | 按需加载 | 非关键资源 |

#### HTTP/2 Server Push 配置

| 图片 | ; rel=preload; as=image | 低 | 首屏图片 |

#### Service Worker 缓存策略

| Network Only | 仅网络 | 实时数据 | 不缓存 |

#### 请求批处理配置

| 重试次数 | 3次 | 失败重试 | 提高成功率 |

4.2 压缩与格式优化

#### CloudFront 压缩配置

| 字体 | Gzip | 10KB | 15-30% | 轻微 |

#### 压缩算法对比

| 不压缩 | 0% | 无 | 全部 | 已压缩文件 |

#### 图片优化策略

| 元数据清除 | EXIF删除 | 5-15% | 无 | 所有图片 |

#### Lambda@Edge 图片优化规则

| Width参数 | 响应式调整 | 按宽度裁剪 | 按需加载 |

#### 图片处理管道配置

| 5. 元数据 | Strip | 删除 | 减小文件 |

五、地理位置优化

5.1 智能路由策略

#### 地理限制配置

| 无限制 | 全球开放 | 覆盖广 | 成本高 | 通用内容 |

#### 价格等级选择指南

| PriceClass_100 | 北美/欧洲 | 100+ | 最低 | 欧美业务 |

#### 地理流量分析指标

| 单位成本 | 成本/流量 | 优化高成本区域 | <$0.10/GB |

#### 区域成本估算模型

| 澳大利亚 | 大洋洲 | $0.140/GB | 35-65% | 根据需求 |

#### 流量分析优化决策

| 南美高占比 | 成本敏感 | 地理限制 | 节省50% |

5.2 边缘计算优化

#### Lambda@Edge 优化点

| Origin Response | 回源后 | 缓存决策 | 3008MB/30秒 | 中 |

#### Viewer Request 优化策略

| 请求验证 | 检查头部/参数 | 安全防护 | API密钥验证 |

#### Origin Response 安全头部配置

| X-XSS-Protection | 1; mode=block | XSS防护 | 现代浏览器 |

#### 缓存控制优化规则

| 字体 | public, max-age=31536000 | 1年 | 1年 |

#### 设备检测逻辑

| desktop | 桌面 | 完整内容 | 标准体验 |

#### Lambda@Edge 最佳实践

| 缓存结果 | 全局变量 | 减少计算 | 注意内存 |

六、监控与成本分析

6.1 实时成本监控

#### CloudFront 关键指标

| 源站延迟 | OriginLatency | 1分钟 | >1000ms | <500ms |

#### 成本分析维度

| 异常检测 | Cost Anomaly | 实时 | 成本突增 | 快速响应 |

#### 成本趋势分析方法

| 环比增长 | (本月-上月)/上月 | >30% | 预算调整 |

import boto3
import pandas as pd
from datetime import datetime, timedelta

class CloudFrontCostMonitor:
    def __init__(self):
        self.cloudwatch = boto3.client('cloudwatch')
        self.ce = boto3.client('ce')  # Cost Explorer
    
    def get_distribution_metrics(self, distribution_id, start_time, end_time):
        """获取分发指标"""
        metrics = {}
        
        # 获取请求数
        metrics['requests'] = self._get_metric(
            distribution_id,
            'Requests',
            start_time,
            end_time
        )
        
        # 获取字节下载量
        metrics['bytes_downloaded'] = self._get_metric(
            distribution_id,
            'BytesDownloaded',
            start_time,
            end_time
        )
        
        # 获取缓存命中率
        metrics['cache_hit_rate'] = self._get_metric(
            distribution_id,
            'CacheHitRate',
            start_time,
            end_time
        )
        
        # 获取错误率
        metrics['error_rate'] = self._get_metric(
            distribution_id,
            '4xxErrorRate',
            start_time,
            end_time
        )
        
        return metrics
    
    def _get_metric(self, distribution_id, metric_name, start_time, end_time):
        """获取单个指标"""
        response = self.cloudwatch.get_metric_statistics(
            Namespace='AWS/CloudFront',
            MetricName=metric_name,
            Dimensions=[
                {
                    'Name': 'DistributionId',
                    'Value': distribution_id
                }
            ],
            StartTime=start_time,
            EndTime=end_time,
            Period=3600,  # 1小时粒度
            Statistics=['Sum', 'Average'],
            Unit='None'
        )
        
        return response['Datapoints']
    
    def analyze_cost_trends(self, distribution_id, days=30):
        """分析成本趋势"""
        end_date = datetime.now().date()
        start_date = end_date - timedelta(days=days)
        
        response = self.ce.get_cost_and_usage(
            TimePeriod={
                'Start': start_date.strftime('%Y-%m-%d'),
                'End': end_date.strftime('%Y-%m-%d')
            },
            Granularity='DAILY',
            Metrics=['UnblendedCost'],
            Filter={
                'And': [
                    {
                        'Dimensions': {
                            'Key': 'SERVICE',
                            'Values': ['Amazon CloudFront']
                        }
                    },
                    {
                        'Tags': {
                            'Key': 'DistributionId',
                            'Values': [distribution_id]
                        }
                    }
                ]
            },
            GroupBy=[
                {
                    'Type': 'DIMENSION',
                    'Key': 'USAGE_TYPE'
                }
            ]
        )
        
        # 处理成本数据
        cost_data = []
        for result in response['ResultsByTime']:
            date = result['TimePeriod']['Start']
            for group in result['Groups']:
                usage_type = group['Keys'][0]
                cost = float(group['Metrics']['UnblendedCost']['Amount'])
                
                cost_data.append({
                    'date': date,
                    'usage_type': usage_type,
                    'cost': cost
                })
        
        df = pd.DataFrame(cost_data)
        
        # 分析成本组成
        cost_breakdown = df.groupby('usage_type')['cost'].sum().sort_values(ascending=False)
        
        # 识别成本异常
        daily_costs = df.groupby('date')['cost'].sum()
        mean_cost = daily_costs.mean()
        std_cost = daily_costs.std()
        anomalies = daily_costs[daily_costs > mean_cost + 2  std_cost]
        
        return {
            'total_cost': df['cost'].sum(),
            'daily_average': mean_cost,
            'cost_breakdown': cost_breakdown.to_dict(),
            'anomalies': anomalies.to_dict(),
            'trend': 'increasing' if daily_costs.iloc[-7:].mean() > mean_cost else 'stable'
        }
    
    def generate_optimization_report(self, distribution_id):
        """生成优化报告"""
        metrics = self.get_distribution_metrics(
            distribution_id,
            datetime.now() - timedelta(days=7),
            datetime.now()
        )
        
        cost_analysis = self.analyze_cost_trends(distribution_id)
        
        # 计算优化潜力
        cache_hit_rate = sum(m['Average'] for m in metrics['cache_hit_rate']) / len(metrics['cache_hit_rate'])
        potential_savings = (1 - cache_hit_rate)  cost_analysis['total_cost']  0.5  # 假设50%可优化
        
        report = f"""
        CloudFront 优化报告
        ===================
        
        分发 ID: {distribution_id}
        分析周期: 最近30天
        
        成本概况:

总成本: ${cost_analysis['total_cost']:.2f}
日均成本: ${cost_analysis['daily_average']:.2f}
成本趋势: {cost_analysis['trend']}

        
        性能指标:

缓存命中率: {cache_hit_rate:.1%}
错误率: {sum(m['Average'] for m in metrics['error_rate']) / len(metrics['error_rate']):.2%}

        
        优化建议:
        1. 提升缓存命中率可节省: ${potential_savings:.2f}
        2. 主要成本来源: {list(cost_analysis['cost_breakdown'].keys())[0]}
        3. 成本异常日期: {list(cost_analysis['anomalies'].keys())}
        
        行动计划:
        1. 优化缓存策略，将命中率提升至 95%
        2. 启用压缩减少数据传输
        3. 使用 Origin Shield 减少回源
        4. 调整价格等级至 PriceClass_100
        """
        
        return report

class AutoOptimizer:
    def __init__(self, distribution_id):
        self.distribution_id = distribution_id
        self.cloudfront = boto3.client('cloudfront')
        self.monitor = CloudFrontCostMonitor()
    
    def run_optimization(self):
        """运行自动优化"""
        # 1. 分析当前状态
        current_config = self._get_distribution_config()
        metrics = self.monitor.get_distribution_metrics(
            self.distribution_id,
            datetime.now() - timedelta(days=7),
            datetime.now()
        )
        
        # 2. 生成优化建议
        optimizations = []
        
        # 检查缓存配置
        if not current_config.get('DefaultCacheBehavior', {}).get('Compress'):
            optimizations.append({
                'type': 'enable_compression',
                'estimated_savings': '15-25%'
            })
        
        # 检查价格等级
        if current_config.get('PriceClass') == 'PriceClass_All':
            optimizations.append({
                'type': 'adjust_price_class',
                'estimated_savings': '10-20%'
            })
        
        # 检查 Origin Shield
        if not current_config.get('Origins', [{}])[0].get('OriginShield'):
            optimizations.append({
                'type': 'enable_origin_shield',
                'estimated_savings': '20-30%'
            })
        
        # 3. 应用优化
        for optimization in optimizations:
            self._apply_optimization(optimization)
        
        return optimizations
    
    def _apply_optimization(self, optimization):
        """应用单个优化"""
        if optimization['type'] == 'enable_compression':
            self._enable_compression()
        elif optimization['type'] == 'adjust_price_class':
            self._adjust_price_class()
        elif optimization['type'] == 'enable_origin_shield':
            self._enable_origin_shield()
    
    def schedule_optimization_review(self):
        """定期优化审查"""
        import schedule
        
        def review_and_optimize():
            report = self.monitor.generate_optimization_report(self.distribution_id)
            print(report)
            
            # 如果成本超过阈值，自动优化
            cost_analysis = self.monitor.analyze_cost_trends(self.distribution_id)
            if cost_analysis['daily_average'] > 100:  # $100/天阈值
                optimizations = self.run_optimization()
                print(f"Applied {len(optimizations)} optimizations")
        
        # 每周一执行优化审查
        schedule.every().monday.at("09:00").do(review_and_optimize)
        
        # 每天检查成本异常
        schedule.every().day.at("10:00").do(lambda: self._check_cost_anomalies())

错误示例及修正
class CommonMistakes:
    @staticmethod
    def avoid_cache_mistakes():
        """避免缓存配置错误"""
        # ❌ 错误：不区分内容类型
        wrong_config = {
            'CacheBehaviors': [{
                'PathPattern': '',
                'MinTTL': 0,
                'MaxTTL': 86400
            }]
        }
        
        # ✅ 正确：根据内容类型设置不同缓存
        correct_config = {
            'CacheBehaviors': [
                {
                    'PathPattern': '.jpg',
                    'MinTTL': 86400,
                    'MaxTTL': 31536000
                },
                {
                    'PathPattern': '/api/',
                    'MinTTL': 0,
                    'MaxTTL': 0
                }
            ]
        }
        
        return correct_config
    
    @staticmethod
    def avoid_origin_mistakes():
        """避免源站配置错误"""
        # ❌ 错误：没有故障转移
        wrong_origin = {
            'Origins': [{
                'Id': 'single-origin',
                'DomainName': 'origin.example.com'
            }]
        }
        
        # ✅ 正确：配置源站组实现高可用
        correct_origin = {
            'OriginGroups': {
                'Items': [{
                    'Id': 'origin-group',
                    'FailoverCriteria': {
                        'StatusCodes': {
                            'Items': [500, 502, 503, 504]
                        }
                    },
                    'Members': {
                        'Items': [
                            {'OriginId': 'primary'},
                            {'OriginId': 'secondary'}
                        ]
                    }
                }]
            }
        }
        
        return correct_origin

八、ROI 计算工具

class CloudFrontROICalculator:
    def calculate_optimization_roi(self, current_monthly_cost, optimization_plan):
        """计算优化投资回报率"""
        
        # 优化成本
        implementation_costs = {
            'engineering_hours': optimization_plan.get('hours', 40)  150,  # $150/小时
            'lambda_edge': 100,  # Lambda@Edge 月度成本
            'origin_shield': current_monthly_cost  0.05,  # Origin Shield 约5%额外成本
            'monitoring_tools': 50  # 监控工具成本
        }
        
        total_implementation_cost = sum(implementation_costs.values())
        
        # 预期节省
        expected_savings = {
            'cache_optimization': current_monthly_cost  0.25,
            'compression': current_monthly_cost  0.15,
            'geo_optimization': current_monthly_cost  0.10,
            'request_reduction': current_monthly_cost  0.05
        }
        
        monthly_savings = sum(expected_savings.values())
        
        # ROI 计算
        roi_months = total_implementation_cost / monthly_savings if monthly_savings > 0 else float('inf')
        annual_roi = ((monthly_savings  12 - total_implementation_cost) / total_implementation_cost)  100
        
        return {
            'implementation_cost': total_implementation_cost,
            'monthly_savings': monthly_savings,
            'payback_period_months': roi_months,
            'annual_roi_percentage': annual_roi,
            'three_year_savings': monthly_savings  36 - total_implementation_cost
        }

StablePayx 团队专注于 AWS 成本优化，已帮助超过 100 家企业降低云成本。如需专业的 CloudFront 优化服务，请联系我们。*