5.8.3 Serverless 基础 #

// 传统架构示例
package main

import (
    "fmt"
    "net/http"
    "log"
)

type Server struct {
    // 服务器状态
    connections int
    cache       map[string]string
}

func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
    s.connections++
    // 处理请求...
    fmt.Fprintf(w, "Connections: %d", s.connections)
}

func main() {
    server := &Server{
        cache: make(map[string]string),
    }

    http.HandleFunc("/", server.handleRequest)
    log.Fatal(http.ListenAndServe(":8080", nil))
}

// Serverless 函数示例
package main

import (
    "context"
    "encoding/json"
    "fmt"
)

// Event 定义输入事件结构
type Event struct {
    Name    string `json:"name"`
    Message string `json:"message"`
}

// Response 定义响应结构
type Response struct {
    StatusCode int               `json:"statusCode"`
    Headers    map[string]string `json:"headers"`
    Body       string            `json:"body"`
}

// HandleRequest 是无服务器函数的入口点
func HandleRequest(ctx context.Context, event Event) (Response, error) {
    // 无状态处理逻辑
    responseBody := fmt.Sprintf("Hello %s! Message: %s", event.Name, event.Message)

    return Response{
        StatusCode: 200,
        Headers: map[string]string{
            "Content-Type": "application/json",
        },
        Body: responseBody,
    }, nil
}

func main() {
    // 在本地测试
    event := Event{
        Name:    "World",
        Message: "This is a serverless function",
    }

    response, err := HandleRequest(context.Background(), event)
    if err != nil {
        panic(err)
    }

    fmt.Printf("Response: %+v\n", response)
}

// aws-lambda-function.go
package main

import (
    "context"
    "encoding/json"
    "fmt"

    "github.com/aws/aws-lambda-go/events"
    "github.com/aws/aws-lambda-go/lambda"
)

type Product struct {
    ID    int    `json:"id"`
    Name  string `json:"name"`
    Price float64 `json:"price"`
}

func handleAPIGatewayRequest(ctx context.Context, request events.APIGatewayProxyRequest) (events.APIGatewayProxyResponse, error) {
    // 解析路径参数
    productID := request.PathParameters["id"]

    // 模拟数据库查询
    product := Product{
        ID:    1,
        Name:  "Sample Product",
        Price: 99.99,
    }

    responseBody, err := json.Marshal(product)
    if err != nil {
        return events.APIGatewayProxyResponse{
            StatusCode: 500,
            Body:       "Internal Server Error",
        }, err
    }

    return events.APIGatewayProxyResponse{
        StatusCode: 200,
        Headers: map[string]string{
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",
        },
        Body: string(responseBody),
    }, nil
}

func main() {
    lambda.Start(handleAPIGatewayRequest)
}

// gcp-cloud-function.go
package cloudfunction

import (
    "context"
    "encoding/json"
    "fmt"
    "net/http"

    "cloud.google.com/go/functions/metadata"
)

type PubSubMessage struct {
    Data []byte `json:"data"`
}

type EventData struct {
    UserID  string `json:"user_id"`
    Action  string `json:"action"`
    Details string `json:"details"`
}

// ProcessEvent 处理 Pub/Sub 事件
func ProcessEvent(ctx context.Context, m PubSubMessage) error {
    meta, err := metadata.FromContext(ctx)
    if err != nil {
        return fmt.Errorf("metadata.FromContext: %v", err)
    }

    var eventData EventData
    if err := json.Unmarshal(m.Data, &eventData); err != nil {
        return fmt.Errorf("json.Unmarshal: %v", err)
    }

    fmt.Printf("Processing event %s for user %s: %s\n",
        meta.EventID, eventData.UserID, eventData.Action)

    // 处理业务逻辑
    return processUserAction(eventData)
}

// HTTPFunction 处理 HTTP 请求
func HTTPFunction(w http.ResponseWriter, r *http.Request) {
    var eventData EventData
    if err := json.NewDecoder(r.Body).Decode(&eventData); err != nil {
        http.Error(w, "Invalid JSON", http.StatusBadRequest)
        return
    }

    if err := processUserAction(eventData); err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }

    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(map[string]string{
        "status": "success",
        "message": "Event processed",
    })
}

func processUserAction(data EventData) error {
    // 实际的业务逻辑处理
    fmt.Printf("Processing action %s for user %s\n", data.Action, data.UserID)
    return nil
}

// event-types.go - 定义各种事件类型
package main

import (
    "encoding/json"
    "time"
)

// HTTPEvent HTTP 请求事件
type HTTPEvent struct {
    Method  string            `json:"method"`
    Path    string            `json:"path"`
    Headers map[string]string `json:"headers"`
    Body    string            `json:"body"`
}

// DatabaseEvent 数据库变更事件
type DatabaseEvent struct {
    Table     string      `json:"table"`
    Operation string      `json:"operation"` // INSERT, UPDATE, DELETE
    OldData   interface{} `json:"old_data,omitempty"`
    NewData   interface{} `json:"new_data,omitempty"`
    Timestamp time.Time   `json:"timestamp"`
}

// FileEvent 文件系统事件
type FileEvent struct {
    Bucket    string    `json:"bucket"`
    Key       string    `json:"key"`
    EventType string    `json:"event_type"` // created, updated, deleted
    Size      int64     `json:"size"`
    Timestamp time.Time `json:"timestamp"`
}

// ScheduledEvent 定时任务事件
type ScheduledEvent struct {
    ScheduleExpression string    `json:"schedule_expression"`
    Time              time.Time `json:"time"`
}

// MessageEvent 消息队列事件
type MessageEvent struct {
    Source    string                 `json:"source"`
    MessageID string                 `json:"message_id"`
    Data      map[string]interface{} `json:"data"`
    Timestamp time.Time              `json:"timestamp"`
}

// event-handler.go - 通用事件处理器
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
)

// EventHandler 定义事件处理器接口
type EventHandler interface {
    Handle(ctx context.Context, event interface{}) error
    EventType() string
}

// UserRegistrationHandler 用户注册事件处理器
type UserRegistrationHandler struct{}

func (h *UserRegistrationHandler) EventType() string {
    return "user.registered"
}

func (h *UserRegistrationHandler) Handle(ctx context.Context, event interface{}) error {
    // 类型断言获取具体事件数据
    data, ok := event.(map[string]interface{})
    if !ok {
        return fmt.Errorf("invalid event data type")
    }

    userID, ok := data["user_id"].(string)
    if !ok {
        return fmt.Errorf("missing user_id in event")
    }

    email, ok := data["email"].(string)
    if !ok {
        return fmt.Errorf("missing email in event")
    }

    // 处理用户注册逻辑
    if err := h.sendWelcomeEmail(userID, email); err != nil {
        return fmt.Errorf("failed to send welcome email: %v", err)
    }

    if err := h.createUserProfile(userID); err != nil {
        return fmt.Errorf("failed to create user profile: %v", err)
    }

    log.Printf("Successfully processed user registration for %s", userID)
    return nil
}

func (h *UserRegistrationHandler) sendWelcomeEmail(userID, email string) error {
    // 发送欢迎邮件的逻辑
    log.Printf("Sending welcome email to %s", email)
    return nil
}

func (h *UserRegistrationHandler) createUserProfile(userID string) error {
    // 创建用户档案的逻辑
    log.Printf("Creating user profile for %s", userID)
    return nil
}

// OrderProcessingHandler 订单处理事件处理器
type OrderProcessingHandler struct{}

func (h *OrderProcessingHandler) EventType() string {
    return "order.created"
}

func (h *OrderProcessingHandler) Handle(ctx context.Context, event interface{}) error {
    data, ok := event.(map[string]interface{})
    if !ok {
        return fmt.Errorf("invalid event data type")
    }

    orderID, ok := data["order_id"].(string)
    if !ok {
        return fmt.Errorf("missing order_id in event")
    }

    // 处理订单逻辑
    if err := h.validateInventory(orderID); err != nil {
        return fmt.Errorf("inventory validation failed: %v", err)
    }

    if err := h.processPayment(orderID); err != nil {
        return fmt.Errorf("payment processing failed: %v", err)
    }

    if err := h.updateOrderStatus(orderID, "processing"); err != nil {
        return fmt.Errorf("failed to update order status: %v", err)
    }

    log.Printf("Successfully processed order %s", orderID)
    return nil
}

func (h *OrderProcessingHandler) validateInventory(orderID string) error {
    log.Printf("Validating inventory for order %s", orderID)
    return nil
}

func (h *OrderProcessingHandler) processPayment(orderID string) error {
    log.Printf("Processing payment for order %s", orderID)
    return nil
}

func (h *OrderProcessingHandler) updateOrderStatus(orderID, status string) error {
    log.Printf("Updating order %s status to %s", orderID, status)
    return nil
}

// EventRouter 事件路由器
type EventRouter struct {
    handlers map[string]EventHandler
}

func NewEventRouter() *EventRouter {
    return &EventRouter{
        handlers: make(map[string]EventHandler),
    }
}

func (r *EventRouter) RegisterHandler(handler EventHandler) {
    r.handlers[handler.EventType()] = handler
}

func (r *EventRouter) RouteEvent(ctx context.Context, eventType string, eventData interface{}) error {
    handler, exists := r.handlers[eventType]
    if !exists {
        return fmt.Errorf("no handler registered for event type: %s", eventType)
    }

    return handler.Handle(ctx, eventData)
}

// 主函数示例
func main() {
    router := NewEventRouter()

    // 注册事件处理器
    router.RegisterHandler(&UserRegistrationHandler{})
    router.RegisterHandler(&OrderProcessingHandler{})

    // 模拟事件处理
    ctx := context.Background()

    // 处理用户注册事件
    userEvent := map[string]interface{}{
        "user_id": "user123",
        "email":   "[email protected]",
    }

    if err := router.RouteEvent(ctx, "user.registered", userEvent); err != nil {
        log.Printf("Error processing user registration: %v", err)
    }

    // 处理订单创建事件
    orderEvent := map[string]interface{}{
        "order_id": "order456",
        "user_id":  "user123",
        "amount":   99.99,
    }

    if err := router.RouteEvent(ctx, "order.created", orderEvent); err != nil {
        log.Printf("Error processing order creation: %v", err)
    }
}

// cold-start-optimization.go - 冷启动优化示例
package main

import (
    "context"
    "database/sql"
    "encoding/json"
    "fmt"
    "log"
    "sync"
    "time"

    _ "github.com/lib/pq"
)

// 全局变量在函数实例生命周期内保持
var (
    db     *sql.DB
    dbOnce sync.Once
    cache  = make(map[string]interface{})
    mu     sync.RWMutex
)

// 初始化函数，在冷启动时执行一次
func init() {
    log.Println("Function cold start - initializing...")

    // 预热缓存
    cache["config"] = map[string]string{
        "api_version": "v1",
        "region":      "us-east-1",
    }

    log.Println("Function initialization completed")
}

// 懒加载数据库连接
func getDB() *sql.DB {
    dbOnce.Do(func() {
        var err error
        db, err = sql.Open("postgres", "postgres://user:pass@localhost/db?sslmode=disable")
        if err != nil {
            log.Printf("Failed to connect to database: %v", err)
            return
        }

        // 设置连接池参数
        db.SetMaxOpenConns(5)
        db.SetMaxIdleConns(2)
        db.SetConnMaxLifetime(5 * time.Minute)

        log.Println("Database connection established")
    })
    return db
}

// 缓存辅助函数
func getFromCache(key string) (interface{}, bool) {
    mu.RLock()
    defer mu.RUnlock()
    value, exists := cache[key]
    return value, exists
}

func setCache(key string, value interface{}) {
    mu.Lock()
    defer mu.Unlock()
    cache[key] = value
}

// 优化后的处理函数
func HandleRequest(ctx context.Context, event map[string]interface{}) (map[string]interface{}, error) {
    startTime := time.Now()

    // 尝试从缓存获取数据
    if cached, exists := getFromCache("user_data"); exists {
        log.Printf("Cache hit - response time: %v", time.Since(startTime))
        return map[string]interface{}{
            "data":   cached,
            "cached": true,
        }, nil
    }

    // 缓存未命中，查询数据库
    db := getDB()
    if db == nil {
        return nil, fmt.Errorf("database connection failed")
    }

    // 执行数据库查询
    var userData map[string]interface{}
    // ... 数据库查询逻辑

    // 更新缓存
    setCache("user_data", userData)

    log.Printf("Database query - response time: %v", time.Since(startTime))
    return map[string]interface{}{
        "data":   userData,
        "cached": false,
    }, nil
}

func main() {
    // 本地测试
    event := map[string]interface{}{
        "user_id": "123",
    }

    response, err := HandleRequest(context.Background(), event)
    if err != nil {
        log.Printf("Error: %v", err)
        return
    }

    fmt.Printf("Response: %+v\n", response)
}

// warmup-strategy.go - 函数预热策略
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"
)

type WarmupEvent struct {
    Source string `json:"source"`
    Type   string `json:"type"`
}

type BusinessEvent struct {
    UserID string `json:"user_id"`
    Action string `json:"action"`
    Data   string `json:"data"`
}

// 检查是否为预热事件
func isWarmupEvent(event interface{}) bool {
    if eventMap, ok := event.(map[string]interface{}); ok {
        if source, exists := eventMap["source"]; exists {
            return source == "warmup"
        }
    }
    return false
}

// 预热处理逻辑
func handleWarmup(ctx context.Context) error {
    log.Println("Handling warmup event")

    // 预热数据库连接
    db := getDB()
    if db != nil {
        // 执行简单查询保持连接活跃
        _, err := db.ExecContext(ctx, "SELECT 1")
        if err != nil {
            log.Printf("Warmup database query failed: %v", err)
        }
    }

    // 预热缓存
    setCache("warmup_time", time.Now())

    log.Println("Warmup completed")
    return nil
}

// 主处理函数
func HandleRequestWithWarmup(ctx context.Context, event interface{}) (interface{}, error) {
    // 检查是否为预热事件
    if isWarmupEvent(event) {
        return map[string]string{"status": "warmed"}, handleWarmup(ctx)
    }

    // 处理业务事件
    return HandleRequest(ctx, event.(map[string]interface{}))
}

// 定时预热函数（可以通过 CloudWatch Events 触发）
func ScheduledWarmup(ctx context.Context, event interface{}) error {
    log.Println("Scheduled warmup triggered")
    return handleWarmup(ctx)
}

// external-state.go - 外部状态管理
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "time"

    "github.com/go-redis/redis/v8"
)

type StateManager struct {
    redis *redis.Client
}

func NewStateManager() *StateManager {
    rdb := redis.NewClient(&redis.Options{
        Addr:     "localhost:6379",
        Password: "",
        DB:       0,
    })

    return &StateManager{
        redis: rdb,
    }
}

func (sm *StateManager) SetState(ctx context.Context, key string, value interface{}, expiration time.Duration) error {
    data, err := json.Marshal(value)
    if err != nil {
        return fmt.Errorf("failed to marshal state: %v", err)
    }

    return sm.redis.Set(ctx, key, data, expiration).Err()
}

func (sm *StateManager) GetState(ctx context.Context, key string, dest interface{}) error {
    data, err := sm.redis.Get(ctx, key).Result()
    if err != nil {
        return fmt.Errorf("failed to get state: %v", err)
    }

    return json.Unmarshal([]byte(data), dest)
}

func (sm *StateManager) DeleteState(ctx context.Context, key string) error {
    return sm.redis.Del(ctx, key).Err()
}

// 使用状态管理的函数示例
func StatefulFunction(ctx context.Context, event map[string]interface{}) (map[string]interface{}, error) {
    sm := NewStateManager()

    userID := event["user_id"].(string)
    stateKey := fmt.Sprintf("user_session:%s", userID)

    // 获取用户会话状态
    var sessionData map[string]interface{}
    err := sm.GetState(ctx, stateKey, &sessionData)
    if err != nil {
        // 首次访问，创建新会话
        sessionData = map[string]interface{}{
            "user_id":    userID,
            "created_at": time.Now(),
            "visit_count": 1,
        }
    } else {
        // 更新访问次数
        visitCount := sessionData["visit_count"].(float64)
        sessionData["visit_count"] = visitCount + 1
        sessionData["last_visit"] = time.Now()
    }

    // 保存状态（30分钟过期）
    if err := sm.SetState(ctx, stateKey, sessionData, 30*time.Minute); err != nil {
        return nil, fmt.Errorf("failed to save session state: %v", err)
    }

    return map[string]interface{}{
        "message": "Session updated",
        "session": sessionData,
    }, nil
}

// structured-logging.go - 结构化日志
package main

import (
    "context"
    "encoding/json"
    "log"
    "os"
    "time"
)

type LogEntry struct {
    Timestamp   time.Time              `json:"timestamp"`
    Level       string                 `json:"level"`
    Message     string                 `json:"message"`
    RequestID   string                 `json:"request_id,omitempty"`
    UserID      string                 `json:"user_id,omitempty"`
    Duration    string                 `json:"duration,omitempty"`
    Error       string                 `json:"error,omitempty"`
    Metadata    map[string]interface{} `json:"metadata,omitempty"`
}

type Logger struct {
    requestID string
    userID    string
}

func NewLogger(requestID, userID string) *Logger {
    return &Logger{
        requestID: requestID,
        userID:    userID,
    }
}

func (l *Logger) log(level, message string, err error, metadata map[string]interface{}) {
    entry := LogEntry{
        Timestamp: time.Now(),
        Level:     level,
        Message:   message,
        RequestID: l.requestID,
        UserID:    l.userID,
        Metadata:  metadata,
    }

    if err != nil {
        entry.Error = err.Error()
    }

    jsonData, _ := json.Marshal(entry)
    log.Println(string(jsonData))
}

func (l *Logger) Info(message string, metadata map[string]interface{}) {
    l.log("INFO", message, nil, metadata)
}

func (l *Logger) Error(message string, err error, metadata map[string]interface{}) {
    l.log("ERROR", message, err, metadata)
}

func (l *Logger) Debug(message string, metadata map[string]interface{}) {
    l.log("DEBUG", message, nil, metadata)
}

// 带监控的函数示例
func MonitoredFunction(ctx context.Context, event map[string]interface{}) (interface{}, error) {
    startTime := time.Now()

    // 从上下文或事件中获取请求ID
    requestID := getRequestID(event)
    userID := getUserID(event)

    logger := NewLogger(requestID, userID)

    logger.Info("Function execution started", map[string]interface{}{
        "event_type": getEventType(event),
    })

    defer func() {
        duration := time.Since(startTime)
        logger.Info("Function execution completed", map[string]interface{}{
            "duration_ms": duration.Milliseconds(),
        })
    }()

    // 业务逻辑处理
    result, err := processBusinessLogic(ctx, event, logger)
    if err != nil {
        logger.Error("Business logic processing failed", err, map[string]interface{}{
            "event": event,
        })
        return nil, err
    }

    return result, nil
}

func getRequestID(event map[string]interface{}) string {
    if id, exists := event["request_id"]; exists {
        return id.(string)
    }
    return "unknown"
}

func getUserID(event map[string]interface{}) string {
    if id, exists := event["user_id"]; exists {
        return id.(string)
    }
    return "anonymous"
}

func getEventType(event map[string]interface{}) string {
    if eventType, exists := event["type"]; exists {
        return eventType.(string)
    }
    return "unknown"
}

func processBusinessLogic(ctx context.Context, event map[string]interface{}, logger *Logger) (interface{}, error) {
    logger.Debug("Processing business logic", map[string]interface{}{
        "step": "validation",
    })

    // 模拟业务处理
    time.Sleep(100 * time.Millisecond)

    return map[string]interface{}{
        "status": "success",
        "data":   "processed",
    }, nil
}

func main() {
    event := map[string]interface{}{
        "request_id": "req-123",
        "user_id":    "user-456",
        "type":       "user_action",
        "data":       "sample data",
    }

    result, err := MonitoredFunction(context.Background(), event)
    if err != nil {
        log.Printf("Function failed: %v", err)
        return
    }

    log.Printf("Function result: %+v", result)
}