Serverless架构

Serverless架构

Serverless将服务器管理交给云厂商，开发者只需关注业务代码，按实际使用量付费。

┌─────────────────────────────────────────────┐
│              Serverless架构                  │
├─────────────┬─────────────┬─────────────────┤
│   FaaS      │    BaaS     │  事件驱动       │
│ (函数计算)  │ (后端服务)  │                 │
├─────────────┼─────────────┼─────────────────┤
│ Lambda      │ DynamoDB    │ API Gateway     │
│ Cloud Func  │ S3          │ EventBridge     │
│ Azure Func  │ Firebase    │ SQS/SNS         │
└─────────────┴─────────────┴─────────────────┘

AWS Lambda示例

# Lambda函数示例
import json
import boto3

def lambda_handler(event, context):
    # 解析请求
    body = json.loads(event.get('body', '{}'))
    user_id = body.get('user_id')
    
    # 调用DynamoDB
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('Users')
    response = table.get_item(Key={'user_id': user_id})
    
    return {
        'statusCode': 200,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps(response.get('Item', {}))
    }

冷启动优化

冷启动是Serverless主要性能瓶颈，通过预热、连接池等策略优化。

# 冷启动优化策略
import boto3

# 复用连接（Lambda执行环境复用）
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Users')

# 预初始化
def init():
    global table
    table = boto3.resource('dynamodb').Table('Users')

# Lambda容器复用
def handler(event, context):
    return {'statusCode': 200, 'body': 'ok'}

Serverless Framework

# serverless.yml配置
service: my-api

provider:
  name: aws
  runtime: python3.9
  region: us-east-1
  memorySize: 256
  timeout: 30

functions:
  getUser:
    handler: handlers/get_user.handler
    events:
      - http:
          path: /users/{id}
          method: get
  createUser:
    handler: handlers/create_user.handler
    events:
      - http:
          path: /users
          method: post

plugins:
  - serverless-python-requirements

厂商锁定规避

# 抽象层设计
# handlers/portable.py
import os

class StorageService:
    """存储服务抽象层"""
    def __init__(self):
        provider = os.environ.get('STORAGE_PROVIDER', 'aws')
        if provider == 'aws':
            from aws_storage import AwsStorage
            self.impl = AwsStorage()
        elif provider == 'gcp':
            from gcp_storage import GcpStorage
            self.impl = GcpStorage()
    
    def put(self, key, data):
        return self.impl.put(key, data)
    
    def get(self, key):
        return self.impl.get(key)

适用场景

适合	不适合
事件驱动	长时间运行
API网关	有状态服务
定时任务	高频低延迟
数据处理	大规模计算