5.8 中间件高级

中间件原理

中间件链式调用

package main

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

// 中间件类型
type Middleware func(http.Handler) http.Handler

// 中间件链
type MiddlewareChain struct {
    middlewares []Middleware
}

func NewMiddlewareChain(middlewares ...Middleware) *MiddlewareChain {
    return &MiddlewareChain{middlewares: middlewares}
}

func (c *MiddlewareChain) Append(middleware Middleware) {
    c.middlewares = append(c.middlewares, middleware)
}

func (c *MiddlewareChain) Then(handler http.Handler) http.Handler {
    // 反向包装，确保执行顺序正确
    for i := len(c.middlewares) - 1; i >= 0; i-- {
        handler = c.middlewares[i](handler)
    }
    return handler
}

func (c *MiddlewareChain) ThenFunc(handler http.HandlerFunc) http.Handler {
    return c.Then(handler)
}

// 示例
func main() {
    chain := NewMiddlewareChain(
        LoggingMiddleware,
        RecoveryMiddleware,
        AuthMiddleware,
    )

    handler := chain.ThenFunc(func(w http.ResponseWriter, r *http.Request) {
        w.Write([]byte("Hello"))
    })

    http.Handle("/", handler)
    http.ListenAndServe(":8080", nil)
}

链路追踪中间件

OpenTelemetry 集成

package middleware

import (
    "context"
    "net/http"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/trace"
)

var tracer = otel.Tracer("http-middleware")

// Trace 中间件
func TraceMiddleware(serviceName string) Middleware {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            ctx := r.Context()

            // 提取 trace context
            propagator := otel.GetTextMapPropagator()
            ctx = propagator.Extract(ctx, propagation.HeaderCarrier(r.Header))

            // 创建 span
            ctx, span := tracer.Start(
                ctx,
                r.URL.Path,
                trace.WithSpanKind(trace.SpanKindServer),
                trace.WithAttributes(
                    attribute.String("http.method", r.Method),
                    attribute.String("http.url", r.URL.String()),
                    attribute.String("http.user_agent", r.UserAgent()),
                ),
            )
            defer span.End()

            // 将 ctx 传递给下游
            r = r.WithContext(ctx)

            // 包装 ResponseWriter 获取状态码
            ww := &wrapWriter{ResponseWriter: w, statusCode: http.StatusOK}
            next.ServeHTTP(ww, r)

            // 设置 span 属性
            span.SetAttributes(
                attribute.Int("http.status_code", ww.statusCode),
            )
        })
    }
}

type wrapWriter struct {
    http.ResponseWriter
    statusCode int
}

func (w *wrapWriter) WriteHeader(code int) {
    w.statusCode = code
    w.ResponseWriter.WriteHeader(code)
}

初始化 OpenTelemetry

package main

import (
    "context"
    "log"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/jaeger"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
)

func InitTracer(serviceName, endpoint string) (*sdktrace.TracerProvider, error) {
    // 创建 Jaeger exporter
    exporter, err := jaeger.New(
        jaeger.WithCollectorEndpoint(
            jaeger.WithEndpoint(endpoint),
        ),
    )
    if err != nil {
        return nil, err
    }

    // 创建 TracerProvider
    tp := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(exporter),
        sdktrace.WithResource(resource.NewWithAttributes(
            semconv.SchemaURL,
            semconv.ServiceNameKey.String(serviceName),
        )),
    )

    otel.SetTracerProvider(tp)
    otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
        propagation.TraceContext{},
        propagation.Baggage{},
    ))

    return tp, nil
}

func main() {
    tp, err := InitTracer("my-service", "http://localhost:14268/api/traces")
    if err != nil {
        log.Fatal(err)
    }
    defer tp.Shutdown(context.Background())

    // 启动服务...
}

限流中间件

令牌桶算法

package middleware

import (
    "net/http"
    "sync"
    "time"

    "golang.org/x/time/rate"
)

// 令牌桶限流
type TokenBucketLimiter struct {
    mu       sync.Mutex
    limiters map[string]*rate.Limiter
    r        rate.Limit
    b        int
}

func NewTokenBucketLimiter(r rate.Limit, b int) *TokenBucketLimiter {
    return &TokenBucketLimiter{
        limiters: make(map[string]*rate.Limiter),
        r:        r,
        b:        b,
    }
}

func (t *TokenBucketLimiter) getLimiter(key string) *rate.Limiter {
    t.mu.Lock()
    defer t.mu.Unlock()

    limiter, ok := t.limiters[key]
    if !ok {
        limiter = rate.NewLimiter(t.r, t.b)
        t.limiters[key] = limiter
    }
    return limiter
}

func (t *TokenBucketLimiter) Allow(key string) bool {
    return t.getLimiter(key).Allow()
}

// 限流中间件 (按 IP)
func RateLimitMiddleware(limiter *TokenBucketLimiter) Middleware {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            ip := getClientIP(r)

            if !limiter.Allow(ip) {
                http.Error(w, "rate limit exceeded", http.StatusTooManyRequests)
                return
            }

            next.ServeHTTP(w, r)
        })
    }
}

// 获取客户端 IP
func getClientIP(r *http.Request) string {
    // 检查 X-Forwarded-For
    if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
        return xff
    }
    // 检查 X-Real-IP
    if xri := r.Header.Get("X-Real-IP"); xri != "" {
        return xri
    }
    // 返回 RemoteAddr
    return r.RemoteAddr
}

滑动窗口限流

package middleware

import (
    "sync"
    "time"
)

type SlidingWindowLimiter struct {
    mu         sync.Mutex
    windows    map[string][]time.Time
    maxReqs    int
    windowSize time.Duration
}

func NewSlidingWindowLimiter(maxReqs int, windowSize time.Duration) *SlidingWindowLimiter {
    return &SlidingWindowLimiter{
        windows:    make(map[string][]time.Time),
        maxReqs:    maxReqs,
        windowSize: windowSize,
    }
}

func (s *SlidingWindowLimiter) Allow(key string) bool {
    s.mu.Lock()
    defer s.mu.Unlock()

    now := time.Now()
    cutoff := now.Add(-s.windowSize)

    // 获取窗口内的请求
    times := s.windows[key]
    valid := []time.Time{}
    for _, t := range times {
        if t.After(cutoff) {
            valid = append(valid, t)
        }
    }

    // 检查是否超过限制
    if len(valid) >= s.maxReqs {
        s.windows[key] = valid
        return false
    }

    // 添加新请求
    valid = append(valid, now)
    s.windows[key] = valid
    return true
}

// 定期清理过期数据
func (s *SlidingWindowLimiter) Cleanup(interval time.Duration) {
    go func() {
        for range time.Tick(interval) {
            s.mu.Lock()
            now := time.Now()
            cutoff := now.Add(-s.windowSize)

            for key, times := range s.windows {
                valid := []time.Time{}
                for _, t := range times {
                    if t.After(cutoff) {
                        valid = append(valid, t)
                    }
                }
                if len(valid) == 0 {
                    delete(s.windows, key)
                } else {
                    s.windows[key] = valid
                }
            }
            s.mu.Unlock()
        }
    }()
}

熔断中间件

熔断器实现

package middleware

import (
    "errors"
    "net/http"
    "sync"
    "time"
)

type State int

const (
    StateClosed State = iota  // 正常状态
    StateOpen                 // 熔断打开
    StateHalfOpen             // 半开状态
)

type CircuitBreaker struct {
    mu           sync.Mutex
    state        State
    failures     int
    maxFailures  int
    resetTimeout time.Duration
    lastFailTime time.Time
    successCount int
}

func NewCircuitBreaker(maxFailures int, resetTimeout time.Duration) *CircuitBreaker {
    return &CircuitBreaker{
        state:        StateClosed,
        maxFailures:  maxFailures,
        resetTimeout: resetTimeout,
    }
}

func (cb *CircuitBreaker) Allow() error {
    cb.mu.Lock()
    defer cb.mu.Unlock()

    switch cb.state {
    case StateOpen:
        // 检查是否应该切换到半开状态
        if time.Since(cb.lastFailTime) > cb.resetTimeout {
            cb.state = StateHalfOpen
            cb.successCount = 0
            return nil
        }
        return errors.New("circuit breaker is open")

    case StateHalfOpen:
        return nil

    default:
        return nil
    }
}

func (cb *CircuitBreaker) RecordSuccess() {
    cb.mu.Lock()
    defer cb.mu.Unlock()

    if cb.state == StateHalfOpen {
        cb.successCount++
        if cb.successCount >= 3 {
            cb.state = StateClosed
            cb.failures = 0
        }
    }
}

func (cb *CircuitBreaker) RecordFailure() {
    cb.mu.Lock()
    defer cb.mu.Unlock()

    cb.failures++
    cb.lastFailTime = time.Now()

    if cb.failures >= cb.maxFailures {
        cb.state = StateOpen
    }
}

// 熔断中间件
func CircuitBreakerMiddleware(cb *CircuitBreaker) Middleware {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            if err := cb.Allow(); err != nil {
                http.Error(w, "service unavailable", http.StatusServiceUnavailable)
                return
            }

            // 包装 ResponseWriter
            ww := &wrapWriter{ResponseWriter: w, statusCode: http.StatusOK}

            next.ServeHTTP(ww, r)

            // 根据状态码记录成功或失败
            if ww.statusCode >= 500 {
                cb.RecordFailure()
            } else {
                cb.RecordSuccess()
            }
        })
    }
}

gobreaker 库使用

package main

import (
    "net/http"
    "time"

    "github.com/sony/gobreaker"
)

var cb *gobreaker.CircuitBreaker

func init() {
    st := gobreaker.Settings{
        Name:        "http-client",
        MaxFailures: 5,
        Timeout:     30 * time.Second,
    }
    cb = gobreaker.NewCircuitBreaker(st)
}

func httpClient(url string) ([]byte, error) {
    _, err := cb.Execute(func() (interface{}, error) {
        resp, err := http.Get(url)
        if err != nil {
            return nil, err
        }
        defer resp.Body.Close()
        // 读取响应...
        return nil, nil
    })
    return nil, err
}

请求日志中间件

Zap 结构化日志

package middleware

import (
    "net/http"
    "time"

    "go.uber.org/zap"
)

func LoggingMiddleware(logger *zap.Logger) Middleware {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            start := time.Now()

            // 包装 ResponseWriter
            ww := &wrapWriter{ResponseWriter: w, statusCode: http.StatusOK, written: 0}

            next.ServeHTTP(ww, r)

            // 记录日志
            logger.Info("HTTP request",
                zap.String("method", r.Method),
                zap.String("path", r.URL.Path),
                zap.String("ip", getClientIP(r)),
                zap.String("user_agent", r.UserAgent()),
                zap.Int("status", ww.statusCode),
                zap.Int("bytes", ww.written),
                zap.Duration("duration", time.Since(start)),
            )
        })
    }
}

type wrapWriter struct {
    http.ResponseWriter
    statusCode int
    written    int
}

func (w *wrapWriter) Write(b []byte) (int, error) {
    n, err := w.ResponseWriter.Write(b)
    w.written += n
    return n, err
}

func (w *wrapWriter) WriteHeader(code int) {
    w.statusCode = code
    w.ResponseWriter.WriteHeader(code)
}

认证中间件

JWT 认证

package middleware

import (
    "context"
    "net/http"
    "strings"

    "github.com/golang-jwt/jwt/v5"
)

type contextKey string

const UserClaimsKey contextKey = "user_claims"

type Claims struct {
    UserID string   `json:"user_id"`
    Roles  []string `json:"roles"`
    jwt.RegisteredClaims
}

func JWTMiddleware(secret string) Middleware {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            authHeader := r.Header.Get("Authorization")
            if authHeader == "" {
                http.Error(w, "missing authorization header", http.StatusUnauthorized)
                return
            }

            tokenStr := strings.TrimPrefix(authHeader, "Bearer ")

            claims := &Claims{}
            token, err := jwt.ParseWithClaims(tokenStr, claims, func(token *jwt.Token) (interface{}, error) {
                return []byte(secret), nil
            })

            if err != nil || !token.Valid {
                http.Error(w, "invalid token", http.StatusUnauthorized)
                return
            }

            ctx := context.WithValue(r.Context(), UserClaimsKey, claims)
            next.ServeHTTP(w, r.WithContext(ctx))
        })
    }
}

// 从 context 获取用户
func GetUserFromContext(ctx context.Context) *Claims {
    claims, ok := ctx.Value(UserClaimsKey).(*Claims)
    if !ok {
        return nil
    }
    return claims
}

// 角色授权中间件
func RequireRole(requiredRoles ...string) Middleware {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            claims := GetUserFromContext(r.Context())
            if claims == nil {
                http.Error(w, "unauthorized", http.StatusUnauthorized)
                return
            }

            // 检查角色
            hasRole := false
            for _, reqRole := range requiredRoles {
                for _, userRole := range claims.Roles {
                    if userRole == reqRole {
                        hasRole = true
                        break
                    }
                }
            }

            if !hasRole {
                http.Error(w, "forbidden", http.StatusForbidden)
                return
            }

            next.ServeHTTP(w, r)
        })
    }
}

中间件链管理

条件中间件

package middleware

import "net/http"

// 条件应用中间件
func If(condition bool, middleware Middleware) Middleware {
    return func(next http.Handler) http.Handler {
        if condition {
            return middleware(next)
        }
        return next
    }
}

// 仅在特定路径应用中间件
func ForPath(pathPrefix string, middleware Middleware) Middleware {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            if strings.HasPrefix(r.URL.Path, pathPrefix) {
                return middleware(next).ServeHTTP(w, r)
            }
            return next.ServeHTTP(w, r)
        })
    }
}

// 中间件组合
func Chain(middlewares ...Middleware) Middleware {
    return func(next http.Handler) http.Handler {
        for i := len(middlewares) - 1; i >= 0; i-- {
            next = middlewares[i](next)
        }
        return next
    }
}

中间件检查清单

[ ] 中间件执行顺序正确
[ ] 错误处理适当
[ ] 性能影响最小化
[ ] 链路追踪集成
[ ] 限流策略合理
[ ] 熔断器配置适当
[ ] 日志结构化
[ ] 认证授权分离
[ ] 中间件可测试
[ ] 支持热重载配置

5.8 中间件高级 ​

中间件原理 ​

中间件链式调用 ​

链路追踪中间件 ​

OpenTelemetry 集成 ​

初始化 OpenTelemetry ​

限流中间件 ​

令牌桶算法 ​

滑动窗口限流 ​

熔断中间件 ​

熔断器实现 ​

gobreaker 库使用 ​

请求日志中间件 ​

Zap 结构化日志 ​

认证中间件 ​

JWT 认证 ​

中间件链管理 ​

条件中间件 ​

中间件检查清单 ​

5.8 中间件高级

中间件原理

中间件链式调用

链路追踪中间件

OpenTelemetry 集成

初始化 OpenTelemetry

限流中间件

令牌桶算法

滑动窗口限流

熔断中间件

熔断器实现

gobreaker 库使用

请求日志中间件

Zap 结构化日志

认证中间件

JWT 认证

中间件链管理

条件中间件

中间件检查清单