AWS Lambda性能优化完全指南

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# lambda_function.py - 优化初始化代码
import json
import boto3
import os

# 在全局范围初始化SDK和连接
s3_client = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')

# 预热连接
table = dynamodb.Table(os.environ['TABLE_NAME'])

# 缓存配置
CONFIG = {
    'bucket_name': os.environ['BUCKET_NAME'],
    'max_retries': 3
}

def lambda_handler(event, context):
    """处理Lambda请求"""
    try:
        # 函数逻辑
        result = process_event(event)
        return {
            'statusCode': 200,
            'body': json.dumps(result)
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

def process_event(event):
    """处理事件逻辑"""
    # 业务逻辑
    pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

# warmer.py - Lambda预热函数
import json
import boto3
import time

lambda_client = boto3.client('lambda')

def warmup_handler(event, context):
    """预热Lambda函数"""
    function_names = [
        'my-app-api-handler',
        'my-app-processor',
        'my-app-notifier'
    ]

    for function_name in function_names:
        # 发送预热请求
        try:
            lambda_client.invoke(
                FunctionName=function_name,
                InvocationType='Event',  # 异步调用
                Payload=json.dumps({'warmup': True})
            )
        except Exception as e:
            print(f"Failed to warm up {function_name}: {e}")

    return {
        'statusCode': 200,
        'body': json.dumps({'message': 'Warmup completed'})
    }

# 使用EventBridge定期预热
# 每5分钟执行一次预热

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

// node.js - 使用esbuild优化打包
// build.js
const esbuild = require('esbuild');

esbuild.build({
  entryPoints: ['index.js'],
  bundle: true,
  minify: true,
  platform: 'node',
  target: ['node14'],
  outfile: 'dist/lambda.js',
  external: ['aws-sdk'],
  define: {
    'process.env.NODE_ENV': '"production"'
  }
}).catch(() => process.exit(1));

// package.json
{
  "scripts": {
    "build": "node build.js",
    "deploy": "npm run build && serverless deploy"
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

# memory_optimizer.py - 内存优化工具
import time
import json
import boto3
import statistics

lambda_client = boto3.client('lambda')

def test_memory_performance(function_name):
    """测试不同内存配置的性能"""
    memory_sizes = [128, 256, 512, 1024, 2048, 3008]
    results = []

    for memory_size in memory_sizes:
        # 更新函数配置
        lambda_client.update_function_configuration(
            FunctionName=function_name,
            MemorySize=memory_size
        )

        # 等待配置更新
        time.sleep(5)

        # 执行测试
        durations = []
        for _ in range(10):
            start_time = time.time()

            response = lambda_client.invoke(
                FunctionName=function_name,
                Payload=json.dumps({'test': True})
            )

            # 等待执行完成
            response['Payload'].read()
            end_time = time.time()

            durations.append((end_time - start_time) * 1000)

        avg_duration = statistics.mean(durations)
        results.append({
            'memory_size': memory_size,
            'avg_duration': avg_duration,
            'cost_per_invocation': calculate_cost(memory_size, avg_duration)
        })

    # 找到性价比最优的配置
    optimal_config = min(results, key=lambda x: x['cost_per_invocation'])
    return optimal_config, results

def calculate_cost(memory_mb, duration_ms):
    """计算单次调用成本"""
    # AWS Lambda 定价（示例）
    cost_per_mb_second = 0.0000166667
    duration_seconds = duration_ms / 1000

    return memory_mb * duration_seconds * cost_per_mb_second

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

# adaptive_memory.py - 自适应内存管理
import os
import boto3
import time

class AdaptiveMemoryManager:
    def __init__(self, function_name):
        self.function_name = function_name
        self.lambda_client = boto3.client('lambda')
        self.cloudwatch = boto3.client('cloudwatch')

    def monitor_and_adjust(self):
        """监控性能并调整内存"""
        # 获取性能指标
        metrics = self.get_performance_metrics()

        # 根据指标调整内存
        new_memory = self.calculate_optimal_memory(metrics)

        if self.should_adjust_memory(new_memory):
            self.adjust_memory(new_memory)

    def get_performance_metrics(self):
        """获取性能指标"""
        end_time = int(time.time())
        start_time = end_time - 3600  # 最近1小时

        response = self.cloudwatch.get_metric_statistics(
            Namespace='AWS/Lambda',
            MetricName='Duration',
            Dimensions=[
                {
                    'Name': 'FunctionName',
                    'Value': self.function_name
                }
            ],
            StartTime=start_time,
            EndTime=end_time,
            Period=60,
            Statistics=['Average', 'Maximum']
        )

        return response['Datapoints']

    def calculate_optimal_memory(self, metrics):
        """计算最优内存配置"""
        if not metrics:
            return 512  # 默认值

        # 基于最大执行时间计算内存需求
        max_duration = max(m['Average'] for m in metrics)

        # 根据经验公式调整
        if max_duration < 100:  # 100ms
            return 256
        elif max_duration < 500:
            return 512
        elif max_duration < 1500:
            return 1024
        else:
            return 2048

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

# concurrent_control.py - 并发控制配置
import boto3

lambda_client = boto3.client('lambda')

def configure_reserved_concurrency():
    """配置预留并发"""
    functions = {
        'critical-api': 100,    # 关键API
        'batch-processor': 50,   # 批处理任务
        'notification-sender': 20 # 通知发送
    }

    for function_name, concurrency in functions.items():
        lambda_client.put_function_concurrency(
            FunctionName=function_name,
            ReservedConcurrentExecutions=concurrency
        )
        print(f"Set {concurrency} reserved concurrency for {function_name}")

def configure_provisioned_concurrency():
    """配置预置并发（用于别名）"""
    lambda_client.put_provisioned_concurrency_config(
        FunctionName='my-api-function',
        Qualifier='production',
        ProvisionedConcurrentExecutions=50
    )

# Serverless Framework配置
# serverless.yml
service: my-app

provider:
  name: aws
  runtime: python3.9

functions:
  criticalFunction:
    handler: handler.main
    reservedConcurrency: 10
    provisionedConcurrency: 5
    events:
      - http:
          path: /critical
          method: get

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

# rate_limiter.py - 限流实现
import time
import json
from collections import defaultdict
import threading

class RateLimiter:
    def __init__(self, max_requests, time_window):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = defaultdict(list)
        self.lock = threading.Lock()

    def is_allowed(self, key):
        """检查是否允许请求"""
        with self.lock:
            now = time.time()

            # 清理过期请求
            self.requests[key] = [
                req_time for req_time in self.requests[key]
                if now - req_time < self.time_window
            ]

            # 检查是否超过限制
            if len(self.requests[key]) >= self.max_requests:
                return False

            # 记录新请求
            self.requests[key].append(now)
            return True

# 在Lambda中使用
rate_limiter = RateLimiter(max_requests=100, time_window=60)

def lambda_handler(event, context):
    client_ip = event['requestContext']['identity']['sourceIp']

    if not rate_limiter.is_allowed(client_ip):
        return {
            'statusCode': 429,
            'body': json.dumps({'error': 'Too many requests'})
        }

    # 处理正常请求
    pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

# async_handler.py - 异步处理模式
import asyncio
import aiohttp
import json
from concurrent.futures import ThreadPoolExecutor

# 使用asyncio进行异步处理
async def process_multiple_apis(urls):
    """并发调用多个API"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

async def fetch_url(session, url):
    """获取单个URL"""
    async with session.get(url) as response:
        return await response.json()

def lambda_handler(event, context):
    """Lambda处理函数"""
    urls = [
        'https://api1.example.com/data',
        'https://api2.example.com/data',
        'https://api3.example.com/data'
    ]

    # 运行异步代码
    loop = asyncio.get_event_loop()
    results = loop.run_until_complete(process_multiple_apis(urls))

    return {
        'statusCode': 200,
        'body': json.dumps(results)
    }

# 使用线程池处理CPU密集型任务
def cpu_intensive_task(data):
    """CPU密集型任务"""
    result = sum(i * i for i in range(data))
    return result

def lambda_handler(event, context):
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = []

        for item in event['items']:
            future = executor.submit(cpu_intensive_task, item)
            futures.append(future)

        results = [future.result() for future in futures]

    return {
        'statusCode': 200,
        'body': json.dumps(results)
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

# data_optimization.py - 数据处理优化
import json
import gzip
import io
import csv

# 压缩响应数据
def compress_response(data):
    """压缩响应数据"""
    json_str = json.dumps(data)
    buffer = io.BytesIO()

    with gzip.GzipFile(fileobj=buffer, mode='w') as gz:
        gz.write(json_str.encode('utf-8'))

    return buffer.getvalue()

# 流式处理大文件
def process_large_csv(event, context):
    """流式处理大CSV文件"""
    s3 = boto3.client('s3')

    # 获取S3事件
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']

    # 流式读取
    response = s3.get_object(Bucket=bucket, Key=key)

    results = []
    for line in response['Body'].iter_lines():
        # 逐行处理，避免内存溢出
        row = json.loads(line)
        processed = process_row(row)
        results.append(processed)

    return {
        'processed_rows': len(results)
    }

# 批量处理优化
def batch_process_items(items, batch_size=100):
    """批量处理项目"""
    batches = [items[i:i + batch_size] for i in range(0, len(items), batch_size)]

    for batch in batches:
        process_batch(batch)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

# monitoring.py - 性能监控装饰器
import time
import functools
import boto3
import os

cloudwatch = boto3.client('cloudwatch')

def monitor_performance(metric_name):
    """性能监控装饰器"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.time()

            try:
                result = func(*args, **kwargs)
                success = True
            except Exception as e:
                success = False
                raise
            finally:
                duration = (time.time() - start_time) * 1000

                # 发送指标到CloudWatch
                cloudwatch.put_metric_data(
                    Namespace='Lambda/Performance',
                    MetricData=[
                        {
                            'MetricName': metric_name,
                            'Value': duration,
                            'Unit': 'Milliseconds',
                            'Dimensions': [
                                {
                                    'Name': 'FunctionName',
                                    'Value': os.environ.get('AWS_LAMBDA_FUNCTION_NAME', 'Unknown')
                                }
                            ]
                        },
                        {
                            'MetricName': f'{metric_name}_Success',
                            'Value': 1 if success else 0,
                            'Unit': 'Count',
                            'Dimensions': [
                                {
                                    'Name': 'FunctionName',
                                    'Value': os.environ.get('AWS_LAMBDA_FUNCTION_NAME', 'Unknown')
                                }
                            ]
                        }
                    ]
                )

            return result

        return wrapper
    return decorator

# 使用示例
@monitor_performance('api_request')
def lambda_handler(event, context):
    """API请求处理"""
    # 业务逻辑
    pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

# retry_handler.py - 重试机制
import time
import random
from functools import wraps

def exponential_backoff(max_retries=3, base_delay=0.1, max_delay=2):
    """指数退避重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries:
                        raise

                    delay = min(base_delay * (2 ** attempt) + random.uniform(0, 0.1), max_delay)
                    time.sleep(delay)

        return wrapper
    return decorator

# 断路器模式
class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = 'CLOSED'  # CLOSED, OPEN, HALF_OPEN

    def call(self, func, *args, **kwargs):
        """断路器调用"""
        if self.state == 'OPEN':
            if time.time() - self.last_failure_time > self.timeout:
                self.state = 'HALF_OPEN'
            else:
                raise Exception('Circuit breaker is OPEN')

        try:
            result = func(*args, **kwargs)
            self.reset()
            return result
        except Exception as e:
            self.record_failure()
            raise

    def record_failure(self):
        """记录失败"""
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.failure_count >= self.failure_threshold:
            self.state = 'OPEN'

    def reset(self):
        """重置断路器"""
        self.failure_count = 0
        self.state = 'CLOSED'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

# cost_optimizer.py - 成本优化
import json
import boto3
from datetime import datetime, timedelta

class LambdaCostOptimizer:
    def __init__(self):
        self.ce_client = boto3.client('ce')
        self.lambda_client = boto3.client('lambda')

    def analyze_costs(self):
        """分析Lambda成本"""
        end_date = datetime.now().date()
        start_date = end_date - timedelta(days=30)

        response = self.ce_client.get_cost_and_usage(
            TimePeriod={
                'Start': start_date.isoformat(),
                'End': end_date.isoformat()
            },
            Granularity='DAILY',
            Metrics=['BlendedCost'],
            GroupBy=[
                {
                    'Type': 'DIMENSION',
                    'Key': 'SERVICE'
                }
            ],
            Filter={
                'Dimensions': {
                    'Key': 'SERVICE',
                    'Values': ['AWS Lambda']
                }
            }
        )

        return response['ResultsByTime']

    def optimize_functions(self):
        """优化函数配置"""
        functions = self.get_all_functions()

        for function in functions:
            metrics = self.get_function_metrics(function['FunctionName'])

            if metrics:
                # 根据使用模式优化配置
                optimal_memory = self.calculate_optimal_memory(metrics)
                optimal_timeout = self.calculate_optimal_timeout(metrics)

                self.update_function_config(
                    function['FunctionName'],
                    optimal_memory,
                    optimal_timeout
                )

    def calculate_optimal_memory(self, metrics):
        """计算最优内存配置"""
        avg_duration = metrics.get('avg_duration', 100)
        max_duration = metrics.get('max_duration', 500)

        # 基于执行时间计算内存需求
        if avg_duration < 100:
            return 128
        elif avg_duration < 500:
            return 256
        elif avg_duration < 2000:
            return 512
        else:
            return 1024

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

# image_processor.py - 图像处理Lambda
import boto3
import io
from PIL import Image

s3_client = boto3.client('s3')
s3_resource = boto3.resource('s3')

def resize_image_handler(event, context):
    """图像缩放处理"""
    for record in event['Records']:
        # 解析S3事件
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']

        # 下载图像
        obj = s3_resource.Object(bucket, key)
        image_data = obj.get()['Body'].read()

        # 处理图像
        image = Image.open(io.BytesIO(image_data))

        # 生成多个尺寸
        sizes = [
            ('thumbnail', (150, 150)),
            ('medium', (500, 500)),
            ('large', (1000, 1000))
        ]

        for size_name, size in sizes:
            resized = image.copy()
            resized.thumbnail(size, Image.LANCZOS)

            # 保存到S3
            output_key = f"processed/{size_name}/{key}"
            output_buffer = io.BytesIO()
            resized.save(output_buffer, format='JPEG', quality=85)

            s3_client.put_object(
                Bucket=bucket,
                Key=output_key,
                Body=output_buffer.getvalue(),
                ContentType='image/jpeg'
            )

# 优化版本：使用预分配内存
def optimized_resize_handler(event, context):
    """优化的图像处理"""
    # 预先分配足够的内存
    import os
    os.environ['PYTHONMALLOC'] = 'malloc'

    resize_image_handler(event, context)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

# api_gateway.py - API网关集成优化
import json
import boto3
from decimal import Decimal

dynamodb = boto3.resource('dynamodb')

class DecimalEncoder(json.JSONEncoder):
    """处理Decimal类型"""
    def default(self, obj):
        if isinstance(obj, Decimal):
            return float(obj)
        return super(DecimalEncoder, self).default(obj)

def api_handler(event, context):
    """API处理函数"""
    http_method = event['httpMethod']
    path = event['path']

    # 路由处理
    if path == '/users' and http_method == 'GET':
        return get_users(event)
    elif path == '/users' and http_method == 'POST':
        return create_user(event)
    else:
        return {
            'statusCode': 404,
            'body': json.dumps({'error': 'Not found'})
        }

def get_users(event):
    """获取用户列表"""
    table = dynamodb.Table('users')

    # 分页参数
    limit = int(event.get('queryStringParameters', {}).get('limit', 10))
    last_key = event.get('queryStringParameters', {}).get('last_key')

    scan_kwargs = {'Limit': limit}
    if last_key:
        scan_kwargs['ExclusiveStartKey'] = {'id': last_key}

    response = table.scan(**scan_kwargs)

    return {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*'
        },
        'body': json.dumps(
            {
                'users': response['Items'],
                'last_key': response.get('LastEvaluatedKey', {}).get('id')
            },
            cls=DecimalEncoder
        )
    }