无服务器架构技术解析：定义与核心概念

// AWS Lambda 函数示例
exports.handler = async (event, context) => {
    const name = event.name || 'World';
    
    return {
        statusCode: 200,
        body: JSON.stringify({
            message: `Hello, ${name}!`,
            timestamp: new Date().toISOString()
        })
    };
};

传统服务器：
月费用 = 服务器租用费 × 24小时 × 30天
无论是否有请求，费用固定
 
无服务器：
月费用 = 调用次数 × 执行时间 × 内存使用
无请求时费用为0

# 图片处理函数示例
def handle_image_upload(event, context):
    """
    处理图片上传事件
    """
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
    
    # 下载图片
    image = download_from_s3(bucket, key)
    
    # 生成缩略图
    thumbnail = create_thumbnail(image, size=(200, 200))
    
    # 上传处理后的图片
    upload_to_s3(thumbnail, bucket, f"thumbnail/{key}")
    
    return {
        'statusCode': 200,
        'body': json.dumps('Image processed successfully')
    }

// 定时数据清理函数
exports.cleanupOldData = async (event, context) => {
    const thirtyDaysAgo = new Date();
    thirtyDaysAgo.setDate(thirtyDaysAgo.getDate() - 30);
    
    try {
        const result = await db.collection('logs')
            .deleteMany({ 
                timestamp: { $lt: thirtyDaysAgo } 
            });
        
        console.log(`Deleted ${result.deletedCount} old records`);
        
        return {
            statusCode: 200,
            body: JSON.stringify({
                message: `Successfully deleted ${result.deletedCount} records`
            })
        };
    } catch (error) {
        console.error('Cleanup failed:', error);
        throw error;
    }
};

# 订单处理主函数
def create_order(event, context):
    order_data = json.loads(event['body'])
    
    # 1. 验证订单数据
    if not validate_order_data(order_data):
        return error_response("Invalid order data")
    
    # 2. 检查库存
    stock_result = invoke_lambda('check-stock-function', {
        'product_id': order_data['product_id'],
        'quantity': order_data['quantity']
    })
    
    if not stock_result['available']:
        return error_response("Product out of stock")
    
    # 3. 创建订单记录
    order_id = generate_order_id()
    order_record = {
        'order_id': order_id,
        'user_id': order_data['user_id'],
        'product_id': order_data['product_id'],
        'quantity': order_data['quantity'],
        'status': 'pending_payment',
        'created_at': datetime.now().isoformat()
    }
    
    save_to_database(order_record)
    
    # 4. 触发支付流程
    invoke_lambda_async('process-payment-function', {
        'order_id': order_id,
        'amount': order_data['amount']
    })
    
    return success_response({
        'order_id': order_id,
        'status': 'created',
        'message': 'Order created successfully'
    })

# serverless.yml 配置示例
service: image-processor
 
provider:
  name: aws
  runtime: python3.9
  
functions:
  imageUpload:
    handler: handler.image_upload
    events:
      - s3:
          bucket: user-uploads
          event: s3:ObjectCreated:*
          
  imageProcess:
    handler: handler.image_process
    events:
      - sns: image-processing-queue
      
  imageAI:
    handler: handler.ai_analysis
    timeout: 30
    memorySize: 1024
    events:
      - sqs:
          arn: arn:aws:sqs:region:account:image-ai-queue

# 图像处理函数
def image_process(event, context):
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']
        
        # 下载原始图片
        original_image = download_from_s3(bucket, key)
        
        # 并行处理多个版本
        processing_tasks = [
            create_thumbnail(original_image, (150, 150)),
            create_thumbnail(original_image, (300, 300)),
            compress_image(original_image, quality=85),
            convert_format(original_image, 'webp')
        ]
        
        # 等待所有处理完成
        results = asyncio.gather(*processing_tasks)
        
        # 上传处理后的图片
        for size, processed_image in results:
            upload_to_s3(processed_image, bucket, f"processed/{size}/{key}")
        
        # 触发AI分析
        send_to_sqs('image-ai-queue', {
            'bucket': bucket,
            'key': key,
            'processed_keys': [f"processed/{size}/{key}" for size in ['150x150', '300x300', 'compressed', 'webp']]
        })

用户：创建一个处理用户注册的Lambda函数
 
TRAE AI助手：
 
我将为您创建一个用户注册处理函数，包含数据验证、密码加密和数据库存储功能。

// 自动生成的用户注册函数
const bcrypt = require('bcryptjs');
const { DynamoDB } = require('aws-sdk');
 
const dynamodb = new DynamoDB.DocumentClient();
const USERS_TABLE = process.env.USERS_TABLE;
 
exports.handler = async (event, context) => {
    try {
        const { username, email, password } = JSON.parse(event.body);
        
        // 输入验证
        if (!username || !email || !password) {
            return {
                statusCode: 400,
                body: JSON.stringify({ error: 'Missing required fields' })
            };
        }
        
        // 检查用户是否已存在
        const existingUser = await dynamodb.get({
            TableName: USERS_TABLE,
            Key: { email }
        }).promise();
        
        if (existingUser.Item) {
            return {
                statusCode: 409,
                body: JSON.stringify({ error: 'User already exists' })
            };
        }
        
        // 密码加密
        const hashedPassword = await bcrypt.hash(password, 10);
        
        // 创建用户记录
        const user = {
            email,
            username,
            password: hashedPassword,
            createdAt: new Date().toISOString(),
            updatedAt: new Date().toISOString()
        };
        
        await dynamodb.put({
            TableName: USERS_TABLE,
            Item: user
        }).promise();
        
        return {
            statusCode: 201,
            body: JSON.stringify({
                message: 'User created successfully',
                user: { email, username }
            })
        };
        
    } catch (error) {
        console.error('Registration error:', error);
        return {
            statusCode: 500,
            body: JSON.stringify({ error: 'Internal server error' })
        };
    }
};

// .trae/serverless.json 配置文件
{
    "runtime": "nodejs18.x",
    "environment": {
        "USERS_TABLE": "local-users",
        "REGION": "localhost"
    },
    "services": {
        "dynamodb": {
            "port": 8000,
            "inMemory": true
        },
        "s3": {
            "port": 9000,
            "buckets": ["local-uploads", "local-processed"]
        }
    }
}

# TRAE IDE终端命令示例
$ trae deploy --env production
✓ Building function package...
✓ Uploading to AWS Lambda...
✓ Updating function configuration...
✓ Creating new version: 23
✓ Updating alias: production -> v23
 
Deployment completed successfully!
Function: user-registration
Version: 23
Alias: production
URL: https://api.example.com/users/register

TRAE智能体：检测到您的函数存在冷启动优化空间
 
建议：
1. 减少依赖包大小（当前2.3MB）
2. 使用Lambda Layers共享依赖
3. 调整内存配置到512MB
4. 启用Provisioned Concurrency
 
预计优化效果：
- 冷启动时间减少60%
- 内存使用降低30%
- 成本节省25%

// 输入验证和 sanitization
const validator = require('validator');
const xss = require('xss');
 
exports.secureHandler = async (event, context) => {
    const input = JSON.parse(event.body);
    
    // 输入验证
    if (!validator.isEmail(input.email)) {
        return { statusCode: 400, body: 'Invalid email' };
    }
    
    // XSS防护
    const sanitizedInput = {
        name: xss(input.name),
        email: validator.normalizeEmail(input.email)
    };
    
    // 权限检查
    const hasPermission = await checkUserPermission(event.headers.authorization);
    if (!hasPermission) {
        return { statusCode: 403, body: 'Forbidden' };
    }
    
    // 处理业务逻辑
    return processBusinessLogic(sanitizedInput);
};