第2章：API网关实现

package main

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

func main() {
    // 创建网关
    gateway := NewGateway()

    // 注册路由
    gateway.AddRoute("/api/users/*", "http://localhost:8080", []string{"GET", "POST"})
    gateway.AddRoute("/api/orders/*", "http://localhost:8081", []string{"GET", "POST", "PUT"})
    gateway.AddRoute("/api/products/*", "http://localhost:8082", []string{"GET"})

    // 启动服务器
    server := &http.Server{
        Addr:         ":8000",
        Handler:      gateway,
        ReadTimeout:  10 * time.Second,
        WriteTimeout: 10 * time.Second,
    }

    log.Println("API Gateway listening on :8000")
    if err := server.ListenAndServe(); err != nil {
        log.Fatal(err)
    }
}

package main

import (
    "net/http"
    "sync"
)

// Gateway网关结构
type Gateway struct {
    router       *Router
    middlewares  []Middleware
    loadBalancer *LoadBalancer
    rateLimiter  *RateLimiter
    circuitBreaker *CircuitBreaker
    mu           sync.RWMutex
}

func NewGateway() *Gateway {
    return &Gateway{
        router:       NewRouter(),
        middlewares:  []Middleware{},
        loadBalancer: NewLoadBalancer(),
        rateLimiter:  NewRateLimiter(100), // 100 QPS
        circuitBreaker: NewCircuitBreaker(5, 10*time.Second),
    }
}

// 添加路由
func (g *Gateway) AddRoute(path, upstream string, methods []string) {
    route := &Route{
        Path:     path,
        Upstream: upstream,
        Methods:  methods,
    }
    g.router.AddRoute(route)
}

// 添加中间件
func (g *Gateway) Use(middleware Middleware) {
    g.middlewares = append(g.middlewares, middleware)
}

// 实现http.Handler接口
func (g *Gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    // 1. 执行中间件链
    handler := g.buildHandler()
    handler.ServeHTTP(w, r)
}

// 构建处理器链
func (g *Gateway) buildHandler() http.Handler {
    // 最终处理器：反向代理
    handler := http.HandlerFunc(g.proxyHandler)

    // 逆序添加中间件
    for i := len(g.middlewares) - 1; i >= 0; i-- {
        handler = g.middlewares[i](handler)
    }

    // 添加默认中间件
    handler = g.loggerMiddleware(handler)
    handler = g.authMiddleware(handler)
    handler = g.rateLimitMiddleware(handler)
    handler = g.circuitBreakerMiddleware(handler)

    return handler
}

// 反向代理处理器
func (g *Gateway) proxyHandler(w http.ResponseWriter, r *http.Request) {
    // 1. 路由匹配
    route := g.router.Match(r.Method, r.URL.Path)
    if route == nil {
        http.Error(w, "Route not found", http.StatusNotFound)
        return
    }

    // 2. 选择后端实例（负载均衡）
    backend := g.loadBalancer.Select(route.Upstream)

    // 3. 反向代理
    proxy := NewReverseProxy(backend)
    proxy.ServeHTTP(w, r)
}

package main

import (
    "path"
    "strings"
)

type Route struct {
    Path     string
    Upstream string
    Methods  []string
}

type Router struct {
    routes []*Route
}

func NewRouter() *Router {
    return &Router{
        routes: []*Route{},
    }
}

func (r *Router) AddRoute(route *Route) {
    r.routes = append(r.routes, route)
}

func (r *Router) Match(method, urlPath string) *Route {
    for _, route := range r.routes {
        // 检查HTTP方法
        if !contains(route.Methods, method) {
            continue
        }

        // 检查路径匹配
        if r.pathMatch(route.Path, urlPath) {
            return route
        }
    }
    return nil
}

// 路径匹配（支持通配符）
func (r *Router) pathMatch(pattern, urlPath string) bool {
    // 精确匹配
    if pattern == urlPath {
        return true
    }

    // 前缀匹配（/api/users/* 匹配 /api/users/123）
    if strings.HasSuffix(pattern, "/*") {
        prefix := strings.TrimSuffix(pattern, "/*")
        return strings.HasPrefix(urlPath, prefix)
    }

    // 参数匹配（/api/users/{id}）
    patternParts := strings.Split(pattern, "/")
    pathParts := strings.Split(urlPath, "/")

    if len(patternParts) != len(pathParts) {
        return false
    }

    for i := range patternParts {
        if patternParts[i] != pathParts[i] &&
           !strings.HasPrefix(patternParts[i], "{") {
            return false
        }
    }

    return true
}

func contains(slice []string, item string) bool {
    for _, s := range slice {
        if s == item {
            return true
        }
    }
    return false
}

package main

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

type Middleware func(http.Handler) http.Handler

// 日志中间件
func (g *Gateway) loggerMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        start := time.Now()

        // 包装ResponseWriter以记录状态码
        lrw := &loggingResponseWriter{ResponseWriter: w}

        next.ServeHTTP(lrw, r)

        log.Printf("[%s] %s %s - %d (%s)",
            r.Method,
            r.URL.Path,
            r.RemoteAddr,
            lrw.statusCode,
            time.Since(start),
        )
    })
}

type loggingResponseWriter struct {
    http.ResponseWriter
    statusCode int
}

func (lrw *loggingResponseWriter) WriteHeader(code int) {
    lrw.statusCode = code
    lrw.ResponseWriter.WriteHeader(code)
}

// 鉴权中间件（JWT）
func (g *Gateway) authMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        // 跳过健康检查
        if r.URL.Path == "/health" {
            next.ServeHTTP(w, r)
            return
        }

        // 获取Token
        authHeader := r.Header.Get("Authorization")
        if authHeader == "" {
            http.Error(w, "Missing authorization header", http.StatusUnauthorized)
            return
        }

        // 验证Token
        token := strings.TrimPrefix(authHeader, "Bearer ")
        userID, err := ValidateJWT(token)
        if err != nil {
            http.Error(w, "Invalid token", http.StatusUnauthorized)
            return
        }

        // 添加用户信息到Header
        r.Header.Set("X-User-ID", userID)

        next.ServeHTTP(w, r)
    })
}

// 限流中间件
func (g *Gateway) rateLimitMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        // 获取客户端标识
        userID := r.Header.Get("X-User-ID")
        if userID == "" {
            userID = r.RemoteAddr
        }

        // 限流检查
        if !g.rateLimiter.Allow(userID) {
            http.Error(w, "Too many requests", http.StatusTooManyRequests)
            return
        }

        next.ServeHTTP(w, r)
    })
}

// 熔断中间件
func (g *Gateway) circuitBreakerMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        // 检查熔断器状态
        if !g.circuitBreaker.Allow() {
            http.Error(w, "Service temporarily unavailable", http.StatusServiceUnavailable)
            return
        }

        // 包装ResponseWriter以记录结果
        lrw := &loggingResponseWriter{ResponseWriter: w}
        next.ServeHTTP(lrw, r)

        // 记录成功/失败
        if lrw.statusCode >= 500 {
            g.circuitBreaker.RecordFailure()
        } else {
            g.circuitBreaker.RecordSuccess()
        }
    })
}

package main

import (
    "sync"
    "time"

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

type RateLimiter struct {
    limiters map[string]*rate.Limiter
    mu       sync.RWMutex
    qps      int
}

func NewRateLimiter(qps int) *RateLimiter {
    return &RateLimiter{
        limiters: make(map[string]*rate.Limiter),
        qps:      qps,
    }
}

func (rl *RateLimiter) Allow(key string) bool {
    rl.mu.RLock()
    limiter, exists := rl.limiters[key]
    rl.mu.RUnlock()

    if !exists {
        rl.mu.Lock()
        limiter = rate.NewLimiter(rate.Limit(rl.qps), rl.qps*2) // burst = qps*2
        rl.limiters[key] = limiter
        rl.mu.Unlock()
    }

    return limiter.Allow()
}

package main

import (
    "sync"
    "time"
)

type CircuitState int

const (
    StateClosed CircuitState = iota
    StateOpen
    StateHalfOpen
)

type CircuitBreaker struct {
    maxFailures  int
    timeout      time.Duration
    state        CircuitState
    failures     int
    lastFailTime time.Time
    mu           sync.Mutex
}

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

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

    switch cb.state {
    case StateClosed:
        return true

    case StateOpen:
        // 检查是否可以尝试恢复
        if time.Since(cb.lastFailTime) > cb.timeout {
            cb.state = StateHalfOpen
            return true
        }
        return false

    case StateHalfOpen:
        return true

    default:
        return false
    }
}

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

    if cb.state == StateHalfOpen {
        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
    }
}

package main

import (
    "sync"
)

type LoadBalancer struct {
    backends map[string][]string
    indexes  map[string]int
    mu       sync.RWMutex
}

func NewLoadBalancer() *LoadBalancer {
    lb := &LoadBalancer{
        backends: make(map[string][]string),
        indexes:  make(map[string]int),
    }

    // 配置后端实例
    lb.backends["http://localhost:8080"] = []string{
        "http://localhost:8080",
        "http://localhost:8081",
        "http://localhost:8082",
    }

    return lb
}

// 轮询算法
func (lb *LoadBalancer) Select(upstream string) string {
    lb.mu.Lock()
    defer lb.mu.Unlock()

    backends, exists := lb.backends[upstream]
    if !exists || len(backends) == 0 {
        return upstream
    }

    index := lb.indexes[upstream]
    backend := backends[index]

    // 更新索引
    lb.indexes[upstream] = (index + 1) % len(backends)

    return backend
}

package main

import (
    "io"
    "net/http"
    "net/url"
)

func NewReverseProxy(targetURL string) http.Handler {
    target, _ := url.Parse(targetURL)

    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        // 1. 修改请求
        r.URL.Scheme = target.Scheme
        r.URL.Host = target.Host
        r.Host = target.Host

        // 2. 创建新请求
        proxyReq, err := http.NewRequest(r.Method, r.URL.String(), r.Body)
        if err != nil {
            http.Error(w, err.Error(), http.StatusInternalServerError)
            return
        }

        // 3. 复制Header
        for key, values := range r.Header {
            for _, value := range values {
                proxyReq.Header.Add(key, value)
            }
        }

        // 4. 发送请求
        client := &http.Client{Timeout: 10 * time.Second}
        resp, err := client.Do(proxyReq)
        if err != nil {
            http.Error(w, err.Error(), http.StatusBadGateway)
            return
        }
        defer resp.Body.Close()

        // 5. 复制响应Header
        for key, values := range resp.Header {
            for _, value := range values {
                w.Header().Add(key, value)
            }
        }

        // 6. 写入状态码
        w.WriteHeader(resp.StatusCode)

        // 7. 复制响应Body
        io.Copy(w, resp.Body)
    })
}

package main

import (
    "errors"
    "time"

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

var jwtSecret = []byte("your-secret-key")

func ValidateJWT(tokenString string) (string, error) {
    token, err := jwt.Parse(tokenString, func(token *jwt.Token) (interface{}, error) {
        if _, ok := token.Method.(*jwt.SigningMethodHMAC); !ok {
            return nil, errors.New("invalid signing method")
        }
        return jwtSecret, nil
    })

    if err != nil {
        return "", err
    }

    if claims, ok := token.Claims.(jwt.MapClaims); ok && token.Valid {
        // 检查过期时间
        exp := int64(claims["exp"].(float64))
        if time.Now().Unix() > exp {
            return "", errors.New("token expired")
        }

        userID := claims["user_id"].(string)
        return userID, nil
    }

    return "", errors.New("invalid token")
}

func GenerateJWT(userID string) (string, error) {
    claims := jwt.MapClaims{
        "user_id": userID,
        "exp":     time.Now().Add(24 * time.Hour).Unix(),
        "iat":     time.Now().Unix(),
    }

    token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
    return token.SignedString(jwtSecret)
}

type TokenBucket struct {
    capacity  int       // 桶容量
    tokens    int       // 当前令牌数
    rate      int       // 生成速率（每秒）
    lastTime  time.Time
    mu        sync.Mutex
}

func NewTokenBucket(capacity, rate int) *TokenBucket {
    return &TokenBucket{
        capacity: capacity,
        tokens:   capacity,
        rate:     rate,
        lastTime: time.Now(),
    }
}

func (tb *TokenBucket) Allow() bool {
    tb.mu.Lock()
    defer tb.mu.Unlock()

    // 1. 计算新生成的令牌
    now := time.Now()
    elapsed := now.Sub(tb.lastTime).Seconds()
    newTokens := int(elapsed * float64(tb.rate))

    // 2. 更新令牌数（不超过容量）
    tb.tokens = min(tb.capacity, tb.tokens+newTokens)
    tb.lastTime = now

    // 3. 消耗令牌
    if tb.tokens > 0 {
        tb.tokens--
        return true
    }

    return false
}

type LeakyBucket struct {
    capacity int
    queue    chan struct{}
    rate     time.Duration
}

func NewLeakyBucket(capacity int, rate time.Duration) *LeakyBucket {
    lb := &LeakyBucket{
        capacity: capacity,
        queue:    make(chan struct{}, capacity),
        rate:     rate,
    }

    // 启动漏水协程
    go lb.leak()

    return lb
}

func (lb *LeakyBucket) leak() {
    ticker := time.NewTicker(lb.rate)
    defer ticker.Stop()

    for range ticker.C {
        select {
        case <-lb.queue:
            // 漏掉一个请求
        default:
        }
    }
}

func (lb *LeakyBucket) Allow() bool {
    select {
    case lb.queue <- struct{}{}:
        return true
    default:
        return false
    }
}

type SlidingWindow struct {
    limit      int
    windowSize time.Duration
    requests   []time.Time
    mu         sync.Mutex
}

func NewSlidingWindow(limit int, windowSize time.Duration) *SlidingWindow {
    return &SlidingWindow{
        limit:      limit,
        windowSize: windowSize,
        requests:   []time.Time{},
    }
}

func (sw *SlidingWindow) Allow() bool {
    sw.mu.Lock()
    defer sw.mu.Unlock()

    now := time.Now()

    // 1. 移除过期请求
    cutoff := now.Add(-sw.windowSize)
    validRequests := []time.Time{}
    for _, req := range sw.requests {
        if req.After(cutoff) {
            validRequests = append(validRequests, req)
        }
    }
    sw.requests = validRequests

    // 2. 检查是否超过限制
    if len(sw.requests) >= sw.limit {
        return false
    }

    // 3. 记录请求
    sw.requests = append(sw.requests, now)
    return true
}

type RoundRobinBalancer struct {
    backends []string
    index    int
    mu       sync.Mutex
}

func (rb *RoundRobinBalancer) Select() string {
    rb.mu.Lock()
    defer rb.mu.Unlock()

    backend := rb.backends[rb.index]
    rb.index = (rb.index + 1) % len(rb.backends)
    return backend
}

type WeightedBackend struct {
    URL    string
    Weight int
}

type WeightedRoundRobinBalancer struct {
    backends  []*WeightedBackend
    currentWeights []int
    mu        sync.Mutex
}

func (wrb *WeightedRoundRobinBalancer) Select() string {
    wrb.mu.Lock()
    defer wrb.mu.Unlock()

    maxWeightIndex := 0
    totalWeight := 0

    for i, backend := range wrb.backends {
        wrb.currentWeights[i] += backend.Weight
        totalWeight += backend.Weight

        if wrb.currentWeights[i] > wrb.currentWeights[maxWeightIndex] {
            maxWeightIndex = i
        }
    }

    wrb.currentWeights[maxWeightIndex] -= totalWeight
    return wrb.backends[maxWeightIndex].URL
}

type LeastConnectionsBalancer struct {
    backends    []string
    connections map[string]int
    mu          sync.Mutex
}

func (lcb *LeastConnectionsBalancer) Select() string {
    lcb.mu.Lock()
    defer lcb.mu.Unlock()

    minConnections := int(^uint(0) >> 1) // max int
    selected := ""

    for _, backend := range lcb.backends {
        connections := lcb.connections[backend]
        if connections < minConnections {
            minConnections = connections
            selected = backend
        }
    }

    lcb.connections[selected]++
    return selected
}

func (lcb *LeastConnectionsBalancer) Release(backend string) {
    lcb.mu.Lock()
    defer lcb.mu.Unlock()

    lcb.connections[backend]--
}

type ConsistentHashBalancer struct {
    ring       *hashring.HashRing
    backends   []string
}

func NewConsistentHashBalancer(backends []string) *ConsistentHashBalancer {
    ring := hashring.New(backends)
    return &ConsistentHashBalancer{
        ring:     ring,
        backends: backends,
    }
}

func (chb *ConsistentHashBalancer) Select(key string) string {
    backend, _ := chb.ring.GetNode(key)
    return backend
}

type CircuitBreaker struct {
    maxFailures  int           // 失败次数阈值
    timeout      time.Duration // 熔断超时时间
    state        CircuitState  // 当前状态
    failures     int           // 失败次数
    lastFailTime time.Time     // 最后失败时间
    mu           sync.Mutex
}

type CircuitState int

const (
    StateClosed   CircuitState = iota // 关闭（正常）
    StateOpen                          // 打开（熔断）
    StateHalfOpen                      // 半开（尝试恢复）
)

func (cb *CircuitBreaker) Call(fn func() error) error {
    // 1. 检查是否允许请求
    if !cb.Allow() {
        return errors.New("circuit breaker is open")
    }

    // 2. 执行请求
    err := fn()

    // 3. 记录结果
    if err != nil {
        cb.RecordFailure()
        return err
    }

    cb.RecordSuccess()
    return nil
}

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

    switch cb.state {
    case StateClosed:
        return true

    case StateOpen:
        // 检查是否可以尝试恢复
        if time.Since(cb.lastFailTime) > cb.timeout {
            cb.state = StateHalfOpen
            return true
        }
        return false

    case StateHalfOpen:
        return true

    default:
        return false
    }
}

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

    if cb.state == StateHalfOpen {
        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
    }
}

cb := NewCircuitBreaker(5, 10*time.Second)

err := cb.Call(func() error {
    resp, err := http.Get("http://backend-service/api/data")
    if err != nil {
        return err
    }
    defer resp.Body.Close()

    if resp.StatusCode >= 500 {
        return errors.New("server error")
    }

    return nil
})

if err != nil {
    // 降级处理
    return getDataFromCache()
}