Serverless应用开发实战指南

引言

Serverless架构正在重新定义现代应用开发的方式。它让开发者能够专注于业务逻辑，而无需关心底层基础设施的管理。本文将通过实际案例，详细介绍如何从零开始构建一个完整的Serverless应用，涵盖从架构设计到生产部署的全过程。

Serverless应用架构设计

整体架构概览

我们将构建一个电商订单处理系统，包含以下核心功能：

┌─────────────────────────────────────────────────────────────┐
│                    Serverless电商系统                        │
├─────────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐         │
│  │   用户服务   │  │   商品服务   │  │   订单服务   │         │
│  │             │  │             │  │             │         │
│  │ • 用户注册   │  │ • 商品查询   │  │ • 订单创建   │         │
│  │ • 用户认证   │  │ • 库存管理   │  │ • 支付处理   │         │
│  │ • 用户信息   │  │ • 价格计算   │  │ • 订单状态   │         │
│  └─────────────┘  └─────────────┘  └─────────────┘         │
├─────────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐         │
│  │   API网关    │  │   消息队列   │  │   数据存储   │         │
│  │             │  │             │  │             │         │
│  │ • 路由转发   │  │ • 异步处理   │  │ • 用户数据   │         │
│  │ • 认证授权   │  │ • 事件驱动   │  │ • 商品数据   │         │
│  │ • 限流熔断   │  │ • 消息持久   │  │ • 订单数据   │         │
│  └─────────────┘  └─────────────┘  └─────────────┘         │
└─────────────────────────────────────────────────────────────┘

技术栈选择

# 项目结构
serverless-ecommerce/
├── services/
│   ├── user/
│   │   ├── handler.py
│   │   ├── models.py
│   │   └── utils.py
│   ├── product/
│   │   ├── handler.py
│   │   ├── models.py
│   │   └── utils.py
│   └── order/
│       ├── handler.py
│       ├── models.py
│       └── utils.py
├── shared/
│   ├── database.py
│   ├── auth.py
│   ├── utils.py
│   └── exceptions.py
├── tests/
├── deployment/
│   ├── serverless.yml
│   └── docker-compose.yml
└── requirements.txt

# 技术栈配置
TECH_STACK = {
    'runtime': 'python3.9',
    'framework': 'serverless',
    'database': 'dynamodb',
    'cache': 'redis',
    'message_queue': 'sqs',
    'storage': 'oss',
    'monitoring': 'cloudwatch',
    'api_gateway': 'aliyun_api_gateway'
}

核心服务实现

用户服务

# services/user/handler.py
import json
import hashlib
import jwt
from datetime import datetime, timedelta
from shared.database import get_dynamodb_table
from shared.auth import generate_token, verify_token
from shared.exceptions import ValidationError, AuthenticationError

def register_user(event, context):
    """用户注册"""
    try:
        # 解析请求数据
        body = json.loads(event['body'])
        email = body.get('email')
        password = body.get('password')
        username = body.get('username')
        
        # 数据验证
        if not all([email, password, username]):
            raise ValidationError("Missing required fields")
        
        if not is_valid_email(email):
            raise ValidationError("Invalid email format")
        
        if len(password) < 8:
            raise ValidationError("Password must be at least 8 characters")
        
        # 检查用户是否已存在
        users_table = get_dynamodb_table('users')
        existing_user = users_table.get_item(Key={'email': email})
        
        if 'Item' in existing_user:
            raise ValidationError("User already exists")
        
        # 创建用户
        user_id = generate_user_id()
        password_hash = hash_password(password)
        
        user_data = {
            'user_id': user_id,
            'email': email,
            'username': username,
            'password_hash': password_hash,
            'created_at': datetime.utcnow().isoformat(),
            'status': 'active'
        }
        
        users_table.put_item(Item=user_data)
        
        # 生成访问令牌
        token = generate_token(user_id, email)
        
        return {
            'statusCode': 201,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*'
            },
            'body': json.dumps({
                'user_id': user_id,
                'email': email,
                'username': username,
                'token': token
            })
        }
        
    except ValidationError as e:
        return create_error_response(400, str(e))
    except Exception as e:
        print(f"Registration error: {str(e)}")
        return create_error_response(500, "Internal server error")

def login_user(event, context):
    """用户登录"""
    try:
        body = json.loads(event['body'])
        email = body.get('email')
        password = body.get('password')
        
        if not all([email, password]):
            raise ValidationError("Email and password are required")
        
        # 查找用户
        users_table = get_dynamodb_table('users')
        user_response = users_table.get_item(Key={'email': email})
        
        if 'Item' not in user_response:
            raise AuthenticationError("Invalid credentials")
        
        user = user_response['Item']
        
        # 验证密码
        if not verify_password(password, user['password_hash']):
            raise AuthenticationError("Invalid credentials")
        
        # 检查用户状态
        if user.get('status') != 'active':
            raise AuthenticationError("Account is not active")
        
        # 生成访问令牌
        token = generate_token(user['user_id'], user['email'])
        
        # 更新最后登录时间
        users_table.update_item(
            Key={'email': email},
            UpdateExpression='SET last_login = :timestamp',
            ExpressionAttributeValues={
                ':timestamp': datetime.utcnow().isoformat()
            }
        )
        
        return {
            'statusCode': 200,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*'
            },
            'body': json.dumps({
                'user_id': user['user_id'],
                'email': user['email'],
                'username': user['username'],
                'token': token
            })
        }
        
    except (ValidationError, AuthenticationError) as e:
        return create_error_response(401, str(e))
    except Exception as e:
        print(f"Login error: {str(e)}")
        return create_error_response(500, "Internal server error")

def get_user_profile(event, context):
    """获取用户信息"""
    try:
        # 验证令牌
        token = extract_token_from_headers(event['headers'])
        user_data = verify_token(token)
        
        # 获取用户详细信息
        users_table = get_dynamodb_table('users')
        user_response = users_table.get_item(Key={'email': user_data['email']})
        
        if 'Item' not in user_response:
            raise AuthenticationError("User not found")
        
        user = user_response['Item']
        
        # 返回用户信息（不包含敏感数据）
        profile = {
            'user_id': user['user_id'],
            'email': user['email'],
            'username': user['username'],
            'created_at': user['created_at'],
            'last_login': user.get('last_login')
        }
        
        return {
            'statusCode': 200,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*'
            },
            'body': json.dumps(profile)
        }
        
    except AuthenticationError as e:
        return create_error_response(401, str(e))
    except Exception as e:
        print(f"Profile error: {str(e)}")
        return create_error_response(500, "Internal server error")

# 辅助函数
def generate_user_id():
    """生成用户ID"""
    import uuid
    return str(uuid.uuid4())

def hash_password(password):
    """密码哈希"""
    salt = "your_salt_here"  # 实际应用中应使用随机盐
    return hashlib.sha256((password + salt).encode()).hexdigest()

def verify_password(password, password_hash):
    """验证密码"""
    salt = "your_salt_here"
    return hashlib.sha256((password + salt).encode()).hexdigest() == password_hash

def is_valid_email(email):
    """验证邮箱格式"""
    import re
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return re.match(pattern, email) is not None

def extract_token_from_headers(headers):
    """从请求头中提取令牌"""
    auth_header = headers.get('Authorization', '')
    if auth_header.startswith('Bearer '):
        return auth_header[7:]
    raise AuthenticationError("Missing or invalid authorization header")

def create_error_response(status_code, message):
    """创建错误响应"""
    return {
        'statusCode': status_code,
        'headers': {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*'
        },
        'body': json.dumps({'error': message})
    }

商品服务

# services/product/handler.py
import json
import boto3
from decimal import Decimal
from shared.database import get_dynamodb_table
from shared.auth import verify_token
from shared.exceptions import ValidationError, NotFoundError

def get_products(event, context):
    """获取商品列表"""
    try:
        # 解析查询参数
        query_params = event.get('queryStringParameters') or {}
        category = query_params.get('category')
        limit = int(query_params.get('limit', 20))
        offset = query_params.get('offset', '0')
        
        products_table = get_dynamodb_table('products')
        
        # 构建查询
        if category:
            # 按分类查询
            response = products_table.query(
                IndexName='category-index',
                KeyConditionExpression='category = :category',
                ExpressionAttributeValues={':category': category},
                Limit=limit,
                ExclusiveStartKey=json.loads(offset) if offset != '0' else None
            )
        else:
            # 扫描所有商品
            response = products_table.scan(
                Limit=limit,
                ExclusiveStartKey=json.loads(offset) if offset != '0' else None
            )
        
        # 处理返回数据
        products = []
        for item in response['Items']:
            product = {
                'product_id': item['product_id'],
                'name': item['name'],
                'description': item['description'],
                'price': float(item['price']),
                'category': item['category'],
                'stock': int(item['stock']),
                'image_url': item.get('image_url'),
                'created_at': item['created_at']
            }
            products.append(product)
        
        # 分页信息
        next_offset = None
        if 'LastEvaluatedKey' in response:
            next_offset = json.dumps(response['LastEvaluatedKey'])
        
        return {
            'statusCode': 200,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*'
            },
            'body': json.dumps({
                'products': products,
                'next_offset': next_offset,
                'count': len(products)
            })
        }
        
    except Exception as e:
        print(f"Get products error: {str(e)}")
        return create_error_response(500, "Internal server error")

def get_product_detail(event, context):
    """获取商品详情"""
    try:
        product_id = event['pathParameters']['product_id']
        
        products_table = get_dynamodb_table('products')
        response = products_table.get_item(Key={'product_id': product_id})
        
        if 'Item' not in response:
            raise NotFoundError("Product not found")
        
        item = response['Item']
        product = {
            'product_id': item['product_id'],
            'name': item['name'],
            'description': item['description'],
            'price': float(item['price']),
            'category': item['category'],
            'stock': int(item['stock']),
            'image_url': item.get('image_url'),
            'specifications': item.get('specifications', {}),
            'created_at': item['created_at'],
            'updated_at': item.get('updated_at')
        }
        
        return {
            'statusCode': 200,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*'
            },
            'body': json.dumps(product)
        }
        
    except NotFoundError as e:
        return create_error_response(404, str(e))
    except Exception as e:
        print(f"Get product detail error: {str(e)}")
        return create_error_response(500, "Internal server error")

def update_product_stock(event, context):
    """更新商品库存（内部服务调用）"""
    try:
        body = json.loads(event['body'])
        product_id = body['product_id']
        quantity_change = int(body['quantity_change'])
        
        products_table = get_dynamodb_table('products')
        
        # 原子性更新库存
        try:
            response = products_table.update_item(
                Key={'product_id': product_id},
                UpdateExpression='ADD stock :change',
                ConditionExpression='stock + :change >= :zero',
                ExpressionAttributeValues={
                    ':change': quantity_change,
                    ':zero': 0
                },
                ReturnValues='UPDATED_NEW'
            )
            
            new_stock = int(response['Attributes']['stock'])
            
            return {
                'statusCode': 200,
                'body': json.dumps({
                    'product_id': product_id,
                    'new_stock': new_stock,
                    'change': quantity_change
                })
            }
            
        except products_table.meta.client.exceptions.ConditionalCheckFailedException:
            raise ValidationError("Insufficient stock")
            
    except ValidationError as e:
        return create_error_response(400, str(e))
    except Exception as e:
        print(f"Update stock error: {str(e)}")
        return create_error_response(500, "Internal server error")

def search_products(event, context):
    """商品搜索"""
    try:
        query_params = event.get('queryStringParameters') or {}
        keyword = query_params.get('q', '').strip()
        
        if not keyword:
            raise ValidationError("Search keyword is required")
        
        # 使用ElasticSearch或简单的文本匹配
        products_table = get_dynamodb_table('products')
        
        # 简单实现：扫描并过滤
        response = products_table.scan(
            FilterExpression='contains(#name, :keyword) OR contains(description, :keyword)',
            ExpressionAttributeNames={'#name': 'name'},
            ExpressionAttributeValues={':keyword': keyword}
        )
        
        products = []
        for item in response['Items']:
            product = {
                'product_id': item['product_id'],
                'name': item['name'],
                'description': item['description'],
                'price': float(item['price']),
                'category': item['category'],
                'stock': int(item['stock']),
                'image_url': item.get('image_url')
            }
            products.append(product)
        
        return {
            'statusCode': 200,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*'
            },
            'body': json.dumps({
                'products': products,
                'keyword': keyword,
                'count': len(products)
            })
        }
        
    except ValidationError as e:
        return create_error_response(400, str(e))
    except Exception as e:
        print(f"Search products error: {str(e)}")
        return create_error_response(500, "Internal server error")

订单服务

# services/order/handler.py
import json
import boto3
from datetime import datetime
from decimal import Decimal
from shared.database import get_dynamodb_table
from shared.auth import verify_token
from shared.exceptions import ValidationError, NotFoundError
from shared.utils import generate_order_id, send_message_to_queue

def create_order(event, context):
    """创建订单"""
    try:
        # 验证用户身份
        token = extract_token_from_headers(event['headers'])
        user_data = verify_token(token)
        user_id = user_data['user_id']
        
        # 解析订单数据
        body = json.loads(event['body'])
        items = body.get('items', [])
        shipping_address = body.get('shipping_address', {})
        
        if not items:
            raise ValidationError("Order items are required")
        
        if not shipping_address:
            raise ValidationError("Shipping address is required")
        
        # 验证商品和计算总价
        total_amount = Decimal('0')
        order_items = []
        
        products_table = get_dynamodb_table('products')
        
        for item in items:
            product_id = item['product_id']
            quantity = int(item['quantity'])
            
            if quantity <= 0:
                raise ValidationError(f"Invalid quantity for product {product_id}")
            
            # 获取商品信息
            product_response = products_table.get_item(Key={'product_id': product_id})
            if 'Item' not in product_response:
                raise ValidationError(f"Product {product_id} not found")
            
            product = product_response['Item']
            
            # 检查库存
            if int(product['stock']) < quantity:
                raise ValidationError(f"Insufficient stock for product {product['name']}")
            
            # 计算小计
            price = Decimal(str(product['price']))
            subtotal = price * quantity
            total_amount += subtotal
            
            order_items.append({
                'product_id': product_id,
                'product_name': product['name'],
                'price': price,
                'quantity': quantity,
                'subtotal': subtotal
            })
        
        # 创建订单
        order_id = generate_order_id()
        order_data = {
            'order_id': order_id,
            'user_id': user_id,
            'items': order_items,
            'total_amount': total_amount,
            'shipping_address': shipping_address,
            'status': 'pending',
            'created_at': datetime.utcnow().isoformat(),
            'updated_at': datetime.utcnow().isoformat()
        }
        
        orders_table = get_dynamodb_table('orders')
        orders_table.put_item(Item=order_data)
        
        # 发送订单处理消息到队列
        message = {
            'order_id': order_id,
            'action': 'process_order',
            'items': [{'product_id': item['product_id'], 'quantity': item['quantity']} 
                     for item in order_items]
        }
        send_message_to_queue('order-processing-queue', message)
        
        # 返回订单信息
        response_data = {
            'order_id': order_id,
            'total_amount': float(total_amount),
            'status': 'pending',
            'created_at': order_data['created_at']
        }
        
        return {
            'statusCode': 201,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*'
            },
            'body': json.dumps(response_data)
        }
        
    except ValidationError as e:
        return create_error_response(400, str(e))
    except Exception as e:
        print(f"Create order error: {str(e)}")
        return create_error_response(500, "Internal server error")

def process_order_queue(event, context):
    """处理订单队列消息"""
    try:
        # 处理SQS消息
        for record in event['Records']:
            message_body = json.loads(record['body'])
            order_id = message_body['order_id']
            action = message_body['action']
            
            if action == 'process_order':
                await process_order_payment(order_id, message_body['items'])
            elif action == 'update_inventory':
                await update_inventory(message_body['items'])
            elif action == 'send_notification':
                await send_order_notification(order_id, message_body['notification_type'])
        
        return {'statusCode': 200}
        
    except Exception as e:
        print(f"Process order queue error: {str(e)}")
        # 返回错误会导致消息重新入队
        raise e

async def process_order_payment(order_id, items):
    """处理订单支付"""
    try:
        orders_table = get_dynamodb_table('orders')
        
        # 获取订单信息
        order_response = orders_table.get_item(Key={'order_id': order_id})
        if 'Item' not in order_response:
            raise NotFoundError(f"Order {order_id} not found")
        
        order = order_response['Item']
        
        # 模拟支付处理
        payment_result = await simulate_payment_processing(order)
        
        if payment_result['success']:
            # 支付成功，更新订单状态
            orders_table.update_item(
                Key={'order_id': order_id},
                UpdateExpression='SET #status = :status, payment_id = :payment_id, updated_at = :timestamp',
                ExpressionAttributeNames={'#status': 'status'},
                ExpressionAttributeValues={
                    ':status': 'paid',
                    ':payment_id': payment_result['payment_id'],
                    ':timestamp': datetime.utcnow().isoformat()
                }
            )
            
            # 扣减库存
            for item in items:
                update_product_stock_internal(item['product_id'], -item['quantity'])
            
            # 发送通知消息
            notification_message = {
                'order_id': order_id,
                'action': 'send_notification',
                'notification_type': 'payment_success'
            }
            send_message_to_queue('order-processing-queue', notification_message)
            
        else:
            # 支付失败，更新订单状态
            orders_table.update_item(
                Key={'order_id': order_id},
                UpdateExpression='SET #status = :status, payment_error = :error, updated_at = :timestamp',
                ExpressionAttributeNames={'#status': 'status'},
                ExpressionAttributeValues={
                    ':status': 'payment_failed',
                    ':error': payment_result['error'],
                    ':timestamp': datetime.utcnow().isoformat()
                }
            )
            
    except Exception as e:
        print(f"Process order payment error: {str(e)}")
        raise e

def get_user_orders(event, context):
    """获取用户订单列表"""
    try:
        # 验证用户身份
        token = extract_token_from_headers(event['headers'])
        user_data = verify_token(token)
        user_id = user_data['user_id']
        
        # 查询参数
        query_params = event.get('queryStringParameters') or {}
        limit = int(query_params.get('limit', 20))
        offset = query_params.get('offset', '0')
        
        orders_table = get_dynamodb_table('orders')
        
        # 查询用户订单
        response = orders_table.query(
            IndexName='user-id-index',
            KeyConditionExpression='user_id = :user_id',
            ExpressionAttributeValues={':user_id': user_id},
            ScanIndexForward=False,  # 按创建时间倒序
            Limit=limit,
            ExclusiveStartKey=json.loads(offset) if offset != '0' else None
        )
        
        # 处理返回数据
        orders = []
        for item in response['Items']:
            order = {
                'order_id': item['order_id'],
                'total_amount': float(item['total_amount']),
                'status': item['status'],
                'created_at': item['created_at'],
                'updated_at': item.get('updated_at'),
                'items_count': len(item['items'])
            }
            orders.append(order)
        
        # 分页信息
        next_offset = None
        if 'LastEvaluatedKey' in response:
            next_offset = json.dumps(response['LastEvaluatedKey'])
        
        return {
            'statusCode': 200,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*'
            },
            'body': json.dumps({
                'orders': orders,
                'next_offset': next_offset,
                'count': len(orders)
            })
        }
        
    except Exception as e:
        print(f"Get user orders error: {str(e)}")
        return create_error_response(500, "Internal server error")

def get_order_detail(event, context):
    """获取订单详情"""
    try:
        # 验证用户身份
        token = extract_token_from_headers(event['headers'])
        user_data = verify_token(token)
        user_id = user_data['user_id']
        
        order_id = event['pathParameters']['order_id']
        
        orders_table = get_dynamodb_table('orders')
        response = orders_table.get_item(Key={'order_id': order_id})
        
        if 'Item' not in response:
            raise NotFoundError("Order not found")
        
        order = response['Item']
        
        # 验证订单所有权
        if order['user_id'] != user_id:
            raise NotFoundError("Order not found")
        
        # 格式化返回数据
        order_detail = {
            'order_id': order['order_id'],
            'total_amount': float(order['total_amount']),
            'status': order['status'],
            'created_at': order['created_at'],
            'updated_at': order.get('updated_at'),
            'shipping_address': order['shipping_address'],
            'items': [
                {
                    'product_id': item['product_id'],
                    'product_name': item['product_name'],
                    'price': float(item['price']),
                    'quantity': int(item['quantity']),
                    'subtotal': float(item['subtotal'])
                }
                for item in order['items']
            ],
            'payment_id': order.get('payment_id')
        }
        
        return {
            'statusCode': 200,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*'
            },
            'body': json.dumps(order_detail)
        }
        
    except NotFoundError as e:
        return create_error_response(404, str(e))
    except Exception as e:
        print(f"Get order detail error: {str(e)}")
        return create_error_response(500, "Internal server error")

# 辅助函数
async def simulate_payment_processing(order):
    """模拟支付处理"""
    import random
    import asyncio
    
    # 模拟支付延迟
    await asyncio.sleep(2)
    
    # 90%成功率
    success = random.random() < 0.9
    
    if success:
        return {
            'success': True,
            'payment_id': f"pay_{generate_order_id()}"
        }
    else:
        return {
            'success': False,
            'error': 'Payment processing failed'
        }

def update_product_stock_internal(product_id, quantity_change):
    """内部库存更新"""
    products_table = get_dynamodb_table('products')
    
    try:
        products_table.update_item(
            Key={'product_id': product_id},
            UpdateExpression='ADD stock :change',
            ConditionExpression='stock + :change >= :zero',
            ExpressionAttributeValues={
                ':change': quantity_change,
                ':zero': 0
            }
        )
    except products_table.meta.client.exceptions.ConditionalCheckFailedException:
        print(f"Failed to update stock for product {product_id}: insufficient stock")
        raise ValidationError("Insufficient stock")

共享组件实现

数据库连接

# shared/database.py
import boto3
from botocore.exceptions import ClientError
import os

# 全局DynamoDB资源
dynamodb = None

def get_dynamodb_resource():
    """获取DynamoDB资源"""
    global dynamodb
    
    if dynamodb is None:
        if os.environ.get('AWS_SAM_LOCAL'):
            # 本地开发环境
            dynamodb = boto3.resource(
                'dynamodb',
                endpoint_url='http://localhost:8000',
                region_name='us-east-1',
                aws_access_key_id='dummy',
                aws_secret_access_key='dummy'
            )
        else:
            # 生产环境
            dynamodb = boto3.resource('dynamodb')
    
    return dynamodb

def get_dynamodb_table(table_name):
    """获取DynamoDB表"""
    dynamodb = get_dynamodb_resource()
    
    # 表名前缀（用于多环境隔离）
    env = os.environ.get('STAGE', 'dev')
    full_table_name = f"{env}-{table_name}"
    
    return dynamodb.Table(full_table_name)

def create_tables_if_not_exist():
    """创建表（如果不存在）"""
    dynamodb = get_dynamodb_resource()
    
    tables_config = {
        'users': {
            'KeySchema': [
                {'AttributeName': 'email', 'KeyType': 'HASH'}
            ],
            'AttributeDefinitions': [
                {'AttributeName': 'email', 'AttributeType': 'S'},
                {'AttributeName': 'user_id', 'AttributeType': 'S'}
            ],
            'GlobalSecondaryIndexes': [
                {
                    'IndexName': 'user-id-index',
                    'KeySchema': [
                        {'AttributeName': 'user_id', 'KeyType': 'HASH'}
                    ],
                    'Projection': {'ProjectionType': 'ALL'},
                    'BillingMode': 'PAY_PER_REQUEST'
                }
            ]
        },
        'products': {
            'KeySchema': [
                {'AttributeName': 'product_id', 'KeyType': 'HASH'}
            ],
            'AttributeDefinitions': [
                {'AttributeName': 'product_id', 'AttributeType': 'S'},
                {'AttributeName': 'category', 'AttributeType': 'S'}
            ],
            'GlobalSecondaryIndexes': [
                {
                    'IndexName': 'category-index',
                    'KeySchema': [
                        {'AttributeName': 'category', 'KeyType': 'HASH'}
                    ],
                    'Projection': {'ProjectionType': 'ALL'},
                    'BillingMode': 'PAY_PER_REQUEST'
                }
            ]
        },
        'orders': {
            'KeySchema': [
                {'AttributeName': 'order_id', 'KeyType': 'HASH'}
            ],
            'AttributeDefinitions': [
                {'AttributeName': 'order_id', 'AttributeType': 'S'},
                {'AttributeName': 'user_id', 'AttributeType': 'S'},
                {'AttributeName': 'created_at', 'AttributeType': 'S'}
            ],
            'GlobalSecondaryIndexes': [
                {
                    'IndexName': 'user-id-index',
                    'KeySchema': [
                        {'AttributeName': 'user_id', 'KeyType': 'HASH'},
                        {'AttributeName': 'created_at', 'KeyType': 'RANGE'}
                    ],
                    'Projection': {'ProjectionType': 'ALL'},
                    'BillingMode': 'PAY_PER_REQUEST'
                }
            ]
        }
    }
    
    env = os.environ.get('STAGE', 'dev')
    
    for table_name, config in tables_config.items():
        full_table_name = f"{env}-{table_name}"
        
        try:
            # 检查表是否存在
            dynamodb.meta.client.describe_table(TableName=full_table_name)
            print(f"Table {full_table_name} already exists")
        except ClientError as e:
            if e.response['Error']['Code'] == 'ResourceNotFoundException':
                # 创建表
                print(f"Creating table {full_table_name}")
                
                table_config = {
                    'TableName': full_table_name,
                    'BillingMode': 'PAY_PER_REQUEST',
                    **config
                }
                
                dynamodb.create_table(**table_config)
                
                # 等待表创建完成
                waiter = dynamodb.meta.client.get_waiter('table_exists')
                waiter.wait(TableName=full_table_name)
                
                print(f"Table {full_table_name} created successfully")
            else:
                raise e

认证授权

# shared/auth.py
import jwt
import os
from datetime import datetime, timedelta
from shared.exceptions import AuthenticationError

# JWT配置
JWT_SECRET = os.environ.get('JWT_SECRET', 'your-secret-key')
JWT_ALGORITHM = 'HS256'
JWT_EXPIRATION_HOURS = 24

def generate_token(user_id, email):
    """生成JWT令牌"""
    payload = {
        'user_id': user_id,
        'email': email,
        'exp': datetime.utcnow() + timedelta(hours=JWT_EXPIRATION_HOURS),
        'iat': datetime.utcnow()
    }
    
    token = jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
    return token

def verify_token(token):
    """验证JWT令牌"""
    try:
        payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
        return payload
    except jwt.ExpiredSignatureError:
        raise AuthenticationError("Token has expired")
    except jwt.InvalidTokenError:
        raise AuthenticationError("Invalid token")

def require_auth(func):
    """认证装饰器"""
    def wrapper(event, context):
        try:
            # 提取令牌
            headers = event.get('headers', {})
            auth_header = headers.get('Authorization', '')
            
            if not auth_header.startswith('Bearer '):
                raise AuthenticationError("Missing or invalid authorization header")
            
            token = auth_header[7:]  # 移除 "Bearer " 前缀
            
            # 验证令牌
            user_data = verify_token(token)
            
            # 将用户信息添加到事件中
            event['user'] = user_data
            
            return func(event, context)
            
        except AuthenticationError as e:
            return {
                'statusCode': 401,
                'headers': {
                    'Content-Type': 'application/json',
                    'Access-Control-Allow-Origin': '*'
                },
                'body': json.dumps({'error': str(e)})
            }
    
    return wrapper

def require_admin(func):
    """管理员权限装饰器"""
    @require_auth
    def wrapper(event, context):
        user_data = event['user']
        
        # 检查管理员权限（这里简化处理）
        if not user_data.get('is_admin', False):
            return {
                'statusCode': 403,
                'headers': {
                    'Content-Type': 'application/json',
                    'Access-Control-Allow-Origin': '*'
                },
                'body': json.dumps({'error': 'Admin access required'})
            }
        
        return func(event, context)
    
    return wrapper

工具函数

# shared/utils.py
import json
import boto3
import uuid
from datetime import datetime

def generate_order_id():
    """生成订单ID"""
    timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
    random_suffix = str(uuid.uuid4())[:8]
    return f"ORD{timestamp}{random_suffix}".upper()

def send_message_to_queue(queue_name, message):
    """发送消息到SQS队列"""
    sqs = boto3.client('sqs')
    
    # 获取队列URL
    env = os.environ.get('STAGE', 'dev')
    queue_url = f"https://sqs.{os.environ.get('AWS_REGION', 'us-east-1')}.amazonaws.com/{os.environ.get('AWS_ACCOUNT_ID')}/{env}-{queue_name}"
    
    try:
        response = sqs.send_message(
            QueueUrl=queue_url,
            MessageBody=json.dumps(message)
        )
        return response['MessageId']
    except Exception as e:
        print(f"Failed to send message to queue {queue_name}: {str(e)}")
        raise e

def send_email_notification(to_email, subject, body):
    """发送邮件通知"""
    ses = boto3.client('ses')
    
    try:
        response = ses.send_email(
            Source=os.environ.get('FROM_EMAIL', 'noreply@example.com'),
            Destination={'ToAddresses': [to_email]},
            Message={
                'Subject': {'Data': subject},
                'Body': {'Text': {'Data': body}}
            }
        )
        return response['MessageId']
    except Exception as e:
        print(f"Failed to send email to {to_email}: {str(e)}")
        raise e

def upload_file_to_oss(file_content, file_name, bucket_name):
    """上传文件到对象存储"""
    s3 = boto3.client('s3')
    
    try:
        s3.put_object(
            Bucket=bucket_name,
            Key=file_name,
            Body=file_content
        )
        
        # 返回文件URL
        return f"https://{bucket_name}.s3.amazonaws.com/{file_name}"
    except Exception as e:
        print(f"Failed to upload file {file_name}: {str(e)}")
        raise e

def validate_email(email):
    """验证邮箱格式"""
    import re
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return re.match(pattern, email) is not None

def validate_phone(phone):
    """验证手机号格式"""
    import re
    pattern = r'^1[3-9]\d{9}$'  # 中国手机号格式
    return re.match(pattern, phone) is not None

class ResponseBuilder:
    """响应构建器"""
    
    @staticmethod
    def success(data=None, status_code=200):
        """成功响应"""
        return {
            'statusCode': status_code,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*',
                'Access-Control-Allow-Headers': 'Content-Type,Authorization',
                'Access-Control-Allow-Methods': 'GET,POST,PUT,DELETE,OPTIONS'
            },
            'body': json.dumps(data) if data else ''
        }
    
    @staticmethod
    def error(message, status_code=400):
        """错误响应"""
        return {
            'statusCode': status_code,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*',
                'Access-Control-Allow-Headers': 'Content-Type,Authorization',
                'Access-Control-Allow-Methods': 'GET,POST,PUT,DELETE,OPTIONS'
            },
            'body': json.dumps({'error': message})
        }

部署配置

Serverless Framework配置

# serverless.yml
service: serverless-ecommerce

provider:
  name: aws
  runtime: python3.9
  region: ${opt:region, 'us-east-1'}
  stage: ${opt:stage, 'dev'}
  memorySize: 512
  timeout: 30
  
  environment:
    STAGE: ${self:provider.stage}
    JWT_SECRET: ${env:JWT_SECRET}
    AWS_ACCOUNT_ID: ${env:AWS_ACCOUNT_ID}
    FROM_EMAIL: ${env:FROM_EMAIL}
  
  iamRoleStatements:
    - Effect: Allow
      Action:
        - dynamodb:Query
        - dynamodb:Scan
        - dynamodb:GetItem
        - dynamodb:PutItem
        - dynamodb:UpdateItem
        - dynamodb:DeleteItem
      Resource:
        - "arn:aws:dynamodb:${self:provider.region}:*:table/${self:provider.stage}-*"
        - "arn:aws:dynamodb:${self:provider.region}:*:table/${self:provider.stage}-*/index/*"
    
    - Effect: Allow
      Action:
        - sqs:SendMessage
        - sqs:ReceiveMessage
        - sqs:DeleteMessage
      Resource:
        - "arn:aws:sqs:${self:provider.region}:*:${self:provider.stage}-*"
    
    - Effect: Allow
      Action:
        - ses:SendEmail
      Resource: "*"

functions:
  # 用户服务
  registerUser:
    handler: services/user/handler.register_user
    events:
      - http:
          path: /api/users/register
          method: post
          cors: true
  
  loginUser:
    handler: services/user/handler.login_user
    events:
      - http:
          path: /api/users/login
          method: post
          cors: true
  
  getUserProfile:
    handler: services/user/handler.get_user_profile
    events:
      - http:
          path: /api/users/profile
          method: get
          cors: true
  
  # 商品服务
  getProducts:
    handler: services/product/handler.get_products
    events:
      - http:
          path: /api/products
          method: get
          cors: true
  
  getProductDetail:
    handler: services/product/handler.get_product_detail
    events:
      - http:
          path: /api/products/{product_id}
          method: get
          cors: true
  
  searchProducts:
    handler: services/product/handler.search_products
    events:
      - http:
          path: /api/products/search
          method: get
          cors: true
  
  # 订单服务
  createOrder:
    handler: services/order/handler.create_order
    events:
      - http:
          path: /api/orders
          method: post
          cors: true
  
  getUserOrders:
    handler: services/order/handler.get_user_orders
    events:
      - http:
          path: /api/orders
          method: get
          cors: true
  
  getOrderDetail:
    handler: services/order/handler.get_order_detail
    events:
      - http:
          path: /api/orders/{order_id}
          method: get
          cors: true
  
  processOrderQueue:
    handler: services/order/handler.process_order_queue
    events:
      - sqs:
          arn:
            Fn::GetAtt:
              - OrderProcessingQueue
              - Arn
          batchSize: 10

resources:
  Resources:
    # SQS队列
    OrderProcessingQueue:
      Type: AWS::SQS::Queue
      Properties:
        QueueName: ${self:provider.stage}-order-processing-queue
        VisibilityTimeoutSeconds: 300
        MessageRetentionPeriod: 1209600  # 14天
        
    # DynamoDB表会通过代码自动创建

plugins:
  - serverless-python-requirements
  - serverless-offline

custom:
  pythonRequirements:
    dockerizePip: true
    slim: true

本地开发环境

# docker-compose.yml
version: '3.8'

services:
  dynamodb-local:
    image: amazon/dynamodb-local:latest
    container_name: dynamodb-local
    ports:
      - "8000:8000"
    command: ["-jar", "DynamoDBLocal.jar", "-sharedDb", "-inMemory"]
  
  redis:
    image: redis:7-alpine
    container_name: redis-local
    ports:
      - "6379:6379"
  
  localstack:
    image: localstack/localstack:latest
    container_name: localstack
    ports:
      - "4566:4566"
    environment:
      - SERVICES=sqs,ses,s3
      - DEBUG=1
      - DATA_DIR=/tmp/localstack/data
    volumes:
      - "./tmp/localstack:/tmp/localstack"
      - "/var/run/docker.sock:/var/run/docker.sock"

测试配置

# tests/test_user_service.py
import pytest
import json
from moto import mock_dynamodb
from services.user.handler import register_user, login_user
from shared.database import create_tables_if_not_exist

@mock_dynamodb
class TestUserService:
    
    def setup_method(self):
        """测试前设置"""
        create_tables_if_not_exist()
    
    def test_register_user_success(self):
        """测试用户注册成功"""
        event = {
            'body': json.dumps({
                'email': 'test@example.com',
                'password': 'password123',
                'username': 'testuser'
            })
        }
        context = {}
        
        response = register_user(event, context)
        
        assert response['statusCode'] == 201
        body = json.loads(response['body'])
        assert 'user_id' in body
        assert 'token' in body
        assert body['email'] == 'test@example.com'
    
    def test_register_user_invalid_email(self):
        """测试无效邮箱注册"""
        event = {
            'body': json.dumps({
                'email': 'invalid-email',
                'password': 'password123',
                'username': 'testuser'
            })
        }
        context = {}
        
        response = register_user(event, context)
        
        assert response['statusCode'] == 400
        body = json.loads(response['body'])
        assert 'error' in body
    
    def test_login_user_success(self):
        """测试用户登录成功"""
        # 先注册用户
        register_event = {
            'body': json.dumps({
                'email': 'test@example.com',
                'password': 'password123',
                'username': 'testuser'
            })
        }
        register_user(register_event, {})
        
        # 然后登录
        login_event = {
            'body': json.dumps({
                'email': 'test@example.com',
                'password': 'password123'
            })
        }
        
        response = login_user(login_event, {})
        
        assert response['statusCode'] == 200
        body = json.loads(response['body'])
        assert 'token' in body
        assert body['email'] == 'test@example.com'

# 运行测试
# pytest tests/ -v --cov=services

监控与运维

性能监控

# monitoring/performance_monitor.py
import time
import json
from functools import wraps

class PerformanceMonitor:
    def __init__(self):
        self.metrics = {}
    
    def track_performance(self, func_name):
        """性能跟踪装饰器"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                start_time = time.time()
                
                try:
                    result = func(*args, **kwargs)
                    status = 'success'
                    return result
                except Exception as e:
                    status = 'error'
                    raise e
                finally:
                    end_time = time.time()
                    duration = end_time - start_time
                    
                    # 记录性能指标
                    self.record_metric(func_name, duration, status)
            
            return wrapper
        return decorator
    
    def record_metric(self, func_name, duration, status):
        """记录性能指标"""
        if func_name not in self.metrics:
            self.metrics[func_name] = {
                'total_calls': 0,
                'success_calls': 0,
                'error_calls': 0,
                'total_duration': 0,
                'max_duration': 0,
                'min_duration': float('inf')
            }
        
        metric = self.metrics[func_name]
        metric['total_calls'] += 1
        metric['total_duration'] += duration
        metric['max_duration'] = max(metric['max_duration'], duration)
        metric['min_duration'] = min(metric['min_duration'], duration)
        
        if status == 'success':
            metric['success_calls'] += 1
        else:
            metric['error_calls'] += 1
    
    def get_metrics_summary(self):
        """获取性能指标摘要"""
        summary = {}
        
        for func_name, metric in self.metrics.items():
            if metric['total_calls'] > 0:
                summary[func_name] = {
                    'total_calls': metric['total_calls'],
                    'success_rate': metric['success_calls'] / metric['total_calls'],
                    'avg_duration': metric['total_duration'] / metric['total_calls'],
                    'max_duration': metric['max_duration'],
                    'min_duration': metric['min_duration'] if metric['min_duration'] != float('inf') else 0
                }
        
        return summary

# 全局性能监控器
performance_monitor = PerformanceMonitor()

# 使用示例
@performance_monitor.track_performance('create_order')
def create_order_with_monitoring(event, context):
    return create_order(event, context)

日志管理

# shared/logging_config.py
import logging
import json
import os
from datetime import datetime

class JSONFormatter(logging.Formatter):
    """JSON格式日志格式化器"""
    
    def format(self, record):
        log_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'level': record.levelname,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno
        }
        
        # 添加额外字段
        if hasattr(record, 'user_id'):
            log_entry['user_id'] = record.user_id
        
        if hasattr(record, 'request_id'):
            log_entry['request_id'] = record.request_id
        
        if hasattr(record, 'execution_time'):
            log_entry['execution_time'] = record.execution_time
        
        return json.dumps(log_entry)

def setup_logging():
    """设置日志配置"""
    logger = logging.getLogger()
    
    # 清除现有处理器
    for handler in logger.handlers:
        logger.removeHandler(handler)
    
    # 创建新处理器
    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())
    
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    
    return logger

def log_function_execution(func):
    """函数执行日志装饰器"""
    @wraps(func)
    def wrapper(event, context):
        logger = logging.getLogger()
        
        # 记录函数开始执行
        logger.info(
            f"Function {func.__name__} started",
            extra={
                'request_id': getattr(context, 'aws_request_id', 'unknown'),
                'function_name': func.__name__
            }
        )
        
        start_time = time.time()
        
        try:
            result = func(event, context)
            
            # 记录成功执行
            execution_time = time.time() - start_time
            logger.info(
                f"Function {func.__name__} completed successfully",
                extra={
                    'request_id': getattr(context, 'aws_request_id', 'unknown'),
                    'function_name': func.__name__,
                    'execution_time': execution_time
                }
            )
            
            return result
            
        except Exception as e:
            # 记录执行错误
            execution_time = time.time() - start_time
            logger.error(
                f"Function {func.__name__} failed: {str(e)}",
                extra={
                    'request_id': getattr(context, 'aws_request_id', 'unknown'),
                    'function_name': func.__name__,
                    'execution_time': execution_time,
                    'error': str(e)
                }
            )
            
            raise e
    
    return wrapper

总结

通过本文的实战指南，我们完整地构建了一个Serverless电商应用，涵盖了从架构设计到生产部署的全过程。Serverless架构的优势在于：

1. 快速开发：专注业务逻辑，无需关心基础设施
2. 自动扩缩容：根据负载自动调整资源
3. 按需付费：只为实际使用的资源付费
4. 高可用性：云平台提供的高可用保障
5. 易于维护：简化的运维工作

关键实践要点：

• 模块化设计：将功能拆分为独立的服务
• 事件驱动：充分利用事件驱动架构
• 无状态原则：保持函数的无状态特性
• 错误处理：完善的错误处理和重试机制
• 监控告警：建立完整的监控体系
• 安全考虑：实现认证授权和数据保护

Serverless架构正在成为现代应用开发的重要选择，掌握其设计和实现技能对于开发者来说具有重要意义。随着技术的不断发展，Serverless将在更多场景中发挥重要作用。