8.3 API网关设计与实现¶

package main

import (
    "net/http"
    "net/http/httputil"
    "net/url"

    "github.com/gin-gonic/gin"
)

// 定义后端服务的URL（示例：用户服务部署在localhost:8081）
var userServiceURL, _ = url.Parse("http://localhost:8081")

// 反向代理处理器：将请求转发到目标服务
func reverseProxy(target *url.URL) gin.HandlerFunc {
    // 创建反向代理实例
    proxy := httputil.NewSingleHostReverseProxy(target)
    // （可选）修改请求头（如添加网关标识）
    originalDirector := proxy.Director
    proxy.Director = func(req *http.Request) {
        originalDirector(req)
        // 添加网关标识，便于后端服务识别请求来源
        req.Header.Set("X-Gateway-Name", "go-gateway")
        req.Header.Set("X-Gateway-Version", "v1.0.0")
    }
    // （可选）修改响应头（如统一设置Cache-Control）
    proxy.ModifyResponse = func(resp *http.Response) error {
        resp.Header.Set("Cache-Control", "no-cache")
        return nil
    }
    // 返回Gin处理器
    return func(c *gin.Context) {
        proxy.ServeHTTP(c.Writer, c.Request)
    }
}

func main() {
    // 1. 创建Gin引擎（开发环境用DebugMode，生产环境用ReleaseMode）
    r := gin.Default() // Default()会自动添加Logger和Recovery中间件

    // 2. 配置路由：将/api/v1/user/*转发到用户服务
    r.Any("/api/v1/user/*path", reverseProxy(userServiceURL))

    // 3. 启动网关服务（监听8080端口）
    if err := r.Run(":8080"); err != nil {
        panic("网关启动失败：" + err.Error())
    }
}

package main

import (
    "fmt"
    "time"

    "github.com/gin-gonic/gin"
)

// LoggerMiddleware 日志中间件
func LoggerMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        // 1. 请求前：记录开始时间、客户端IP、请求信息
        startTime := time.Now()
        clientIP := c.ClientIP()
        method := c.Request.Method
        path := c.Request.URL.Path

        // 2. 执行后续中间件和路由处理函数
        c.Next()

        // 3. 请求后：记录耗时、响应码
        costTime := time.Since(startTime)
        statusCode := c.Writer.Status()

        // 4. 输出日志（生产环境建议用日志库如zap，而非fmt）
        fmt.Printf(
            "[GATEWAY LOG] IP:%s, Method:%s, Path:%s, Status:%d, Cost:%s\n",
            clientIP, method, path, statusCode, costTime,
        )
    }
}

func main() {
    // 创建Gin引擎，添加全局日志中间件
    r := gin.New() // 不用Default()，手动添加中间件
    r.Use(LoggerMiddleware()) // 全局日志中间件
    r.Use(gin.Recovery())     // 全局异常恢复中间件（防止panic导致网关崩溃）

    // 路由配置（同4.1）
    r.Any("/api/v1/user/*path", reverseProxy(userServiceURL))

    // 启动服务
    if err := r.Run(":8080"); err != nil {
        panic("网关启动失败：" + err.Error())
    }
}

package main

import (
    "net/http"
    "strings"

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

// 1. 定义JWT密钥（生产环境需用环境变量或配置中心存储，避免硬编码）
const jwtSecret = "your-secret-key-123"

// JWTAuthMiddleware JWT认证中间件
func JWTAuthMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        // 1. 获取Token：从Authorization头中提取（格式：Bearer <token>）
        authHeader := c.GetHeader("Authorization")
        if authHeader == "" {
            c.JSON(http.StatusUnauthorized, gin.H{"code": 401, "msg": "未携带认证Token"})
            c.Abort() // 终止请求，不执行后续处理
            return
        }

        // 2. 解析Token格式
        parts := strings.SplitN(authHeader, " ", 2)
        if len(parts) != 2 || parts[0] != "Bearer" {
            c.JSON(http.StatusUnauthorized, gin.H{"code": 401, "msg": Token格式错误"})
            c.Abort()
            return
        }
        tokenStr := parts[1]

        // 3. 验证Token有效性
        claims := &jwt.MapClaims{} // 存储Token中的自定义字段（如user_id、role）
        token, err := jwt.ParseWithClaims(tokenStr, claims, func(token *jwt.Token) (interface{}, error) {
            return []byte(jwtSecret), nil
        })
        if err != nil || !token.Valid {
            c.JSON(http.StatusUnauthorized, gin.H{"code": 401, "msg": "Token无效或已过期"})
            c.Abort()
            return
        }

        // 4. Token有效：将用户信息存入上下文，供后续路由使用
        c.Set("user_id", (*claims)["user_id"])
        c.Set("role", (*claims)["role"])

        // 5. 执行后续处理
        c.Next()
    }
}

func main() {
    r := gin.New()
    r.Use(LoggerMiddleware())
    r.Use(gin.Recovery())

    // 2. 为需要认证的路由添加JWT中间件
    userGroup := r.Group("/api/v1/user")
    userGroup.Use(JWTAuthMiddleware()) // 仅该分组下的路由需要认证
    {
        userGroup.Any("/*path", reverseProxy(userServiceURL))
    }

    // 3. 无需认证的路由（如登录接口）
    r.POST("/api/v1/login", func(c *gin.Context) {
        // 登录逻辑：验证用户名密码，生成JWT Token返回
        // ...（此处省略登录逻辑）
        token := generateJWT(123, "user") // 假设用户ID=123，角色=user
        c.JSON(http.StatusOK, gin.H{"code": 200, "msg": "登录成功", "token": token})
    })

    r.Run(":8080")
}

// generateJWT 生成JWT Token（辅助函数）
func generateJWT(userID int, role string) string {
    claims := jwt.MapClaims{
        "user_id": userID,
        "role":    role,
        "exp":     jwt.NewNumericDate(time.Now().Add(24 * time.Hour)), // 过期时间：24小时
    }
    token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
    signedToken, _ := token.SignedString([]byte(jwtSecret))
    return signedToken
}

package main

import "github.com/gin-gonic/gin"

// Plugin 插件接口
type Plugin interface {
    Name() string          // 插件名称（唯一标识）
    Init() error           // 插件初始化（如加载配置）
    Handler() gin.HandlerFunc // 插件的Gin处理器（中间件逻辑）
    Enable() bool          // 是否启用插件
}

package main

import (
    "net/http"
    "strings"

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

// JWTPlugin JWT认证插件
type JWTPlugin struct {
    secret string // JWT密钥
    enable bool   // 是否启用
}

// NewJWTPlugin 创建JWT插件实例
func NewJWTPlugin(secret string, enable bool) *JWTPlugin {
    return &JWTPlugin{
        secret: secret,
        enable: enable,
    }
}

// 实现Plugin接口
func (p *JWTPlugin) Name() string {
    return "jwt-auth"
}

func (p *JWTPlugin) Init() error {
    // 初始化逻辑：如验证密钥是否为空
    if p.secret == "" {
        return fmt.Errorf("jwt插件密钥不能为空")
    }
    fmt.Printf("插件[%s]初始化成功\n", p.Name())
    return nil
}

func (p *JWTPlugin) Handler() gin.HandlerFunc {
    return func(c *gin.Context) {
        // 插件未启用，直接放行
        if !p.enable {
            c.Next()
            return
        }

        // JWT认证逻辑（同4.2的JWTAuthMiddleware）
        authHeader := c.GetHeader("Authorization")
        if authHeader == "" {
            c.JSON(http.StatusUnauthorized, gin.H{"code": 401, "msg": "未携带Token"})
            c.Abort()
            return
        }

        parts := strings.SplitN(authHeader, " ", 2)
        if len(parts) != 2 || parts[0] != "Bearer" {
            c.JSON(http.StatusUnauthorized, gin.H{"code": 401, "msg": "Token格式错误"})
            c.Abort()
            return
        }

        claims := &jwt.MapClaims{}
        token, err := jwt.ParseWithClaims(parts[1], claims, func(token *jwt.Token) (interface{}, error) {
            return []byte(p.secret), nil
        })
        if err != nil || !token.Valid {
            c.JSON(http.StatusUnauthorized, gin.H{"code": 401, "msg": "Token无效"})
            c.Abort()
            return
        }

        // 存储用户信息到上下文
        c.Set("user_id", (*claims)["user_id"])
        c.Next()
    }
}

func (p *JWTPlugin) Enable() bool {
    return p.enable
}

package main

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

    "github.com/gin-gonic/gin"
)

// TokenBucket 令牌桶结构体（限流核心）
type TokenBucket struct {
    capacity  int           // 令牌桶容量（最大并发数）
    rate      int           // 令牌生成速率（每秒生成多少令牌）
    tokens    int           // 当前令牌数
    lastReset time.Time     // 上次重置时间
    mu        sync.Mutex    // 互斥锁（保证并发安全）
}

// NewTokenBucket 创建令牌桶
func NewTokenBucket(capacity, rate int) *TokenBucket {
    return &TokenBucket{
        capacity:  capacity,
        rate:      rate,
        tokens:    capacity, // 初始令牌数=容量
        lastReset: time.Now(),
    }
}

// Take 尝试获取一个令牌（返回true表示成功，false表示限流）
func (tb *TokenBucket) Take() bool {
    tb.mu.Lock()
    defer tb.mu.Unlock()

    // 1. 计算从上次到现在应生成的令牌数
    now := time.Now()
    elapsed := now.Sub(tb.lastReset).Seconds()
    newTokens := int(elapsed * float64(tb.rate))
    tb.tokens = min(tb.tokens+newTokens, tb.capacity) // 不超过桶容量
    tb.lastReset = now

    // 2. 尝试获取令牌
    if tb.tokens > 0 {
        tb.tokens--
        return true
    }
    return false
}

func min(a, b int) int {
    if a < b {
        return a
    }
    return b
}

// RateLimitPlugin 限流插件
type RateLimitPlugin struct {
    bucket *TokenBucket // 令牌桶
    enable bool         // 是否启用
}

// NewRateLimitPlugin 创建限流插件实例
func NewRateLimitPlugin(capacity, rate int, enable bool) *RateLimitPlugin {
    return &RateLimitPlugin{
        bucket: NewTokenBucket(capacity, rate),
        enable: enable,
    }
}

// 实现Plugin接口
func (p *RateLimitPlugin) Name() string {
    return "rate-limit"
}

func (p *RateLimitPlugin) Init() error {
    // 初始化逻辑：验证容量和速率是否合法
    if p.bucket.capacity <= 0 || p.bucket.rate <= 0 {
        return fmt.Errorf("限流插件容量和速率必须大于0")
    }
    fmt.Printf("插件[%s]初始化成功（容量：%d，速率：%d/秒）\n", p.Name(), p.bucket.capacity, p.bucket.rate)
    return nil
}

func (p *RateLimitPlugin) Handler() gin.HandlerFunc {
    return func(c *gin.Context) {
        // 插件未启用，直接放行
        if !p.enable {
            c.Next()
            return
        }

        // 尝试获取令牌：失败则返回429（Too Many Requests）
        if !p.bucket.Take() {
            c.JSON(http.StatusTooManyRequests, gin.H{
                "code": 429,
                "msg":  "请求过于频繁，请稍后再试",
            })
            c.Abort()
            return
        }

        c.Next()
    }
}

func (p *RateLimitPlugin) Enable() bool {
    return p.enable
}

package main

import (
    "sync"

    "github.com/gin-gonic/gin"
)

// PluginManager 插件管理器
var PluginManager = &pluginManager{
    plugins: make(map[string]Plugin),
}

type pluginManager struct {
    sync.RWMutex
    plugins map[string]Plugin // 插件名称→插件实例
}

// RegisterPlugin 注册插件
func (m *pluginManager) RegisterPlugin(plugin Plugin) error {
    m.Lock()
    defer m.Unlock()

    // 检查插件是否已注册
    if _, exists := m.plugins[plugin.Name()]; exists {
        return fmt.Errorf("插件[%s]已注册", plugin.Name())
    }

    // 初始化插件
    if err := plugin.Init(); err != nil {
        return fmt.Errorf("插件[%s]初始化失败：%v", plugin.Name(), err)
    }

    m.plugins[plugin.Name()] = plugin
    return nil
}

// GetEnabledPlugins 获取所有启用的插件
func (m *pluginManager) GetEnabledPlugins() []gin.HandlerFunc {
    m.RLock()
    defer m.RUnlock()

    var handlers []gin.HandlerFunc
    for _, plugin := range m.plugins {
        if plugin.Enable() {
            handlers = append(handlers, plugin.Handler())
        }
    }
    return handlers
}

// 5. 网关主函数（集成插件管理器）
func main() {
    // 1. 注册插件
    if err := PluginManager.RegisterPlugin(NewJWTPlugin("your-secret-key-123", true)); err != nil {
        panic(err)
    }
    if err := PluginManager.RegisterPlugin(NewRateLimitPlugin(100, 10, true)); err != nil { // 容量100，速率10/秒
        panic(err)
    }

    // 2. 创建Gin引擎，加载所有启用的插件（中间件）
    r := gin.New()
    r.Use(gin.Recovery())
    r.Use(PluginManager.GetEnabledPlugins()...) // 加载插件中间件
    r.Use(LoggerMiddleware())                  // 加载自定义日志中间件

    // 3. 配置动态路由（同4.3）
    if err := routeManager.LoadConfig("routes.yaml"); err != nil {
        panic(err)
    }
    r.Any("/*path", dynamicRouteHandler)

    // 4. 启动服务
    if err := r.Run(":8080"); err != nil {
        panic("网关启动失败：" + err.Error())
    }
}

package main

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

// HealthChecker 健康检查器
type HealthChecker struct {
    serviceInstances map[string][]string // 服务名→实例列表（同服务发现）
    healthyInstances map[string][]string // 服务名→健康实例列表
    mu               sync.RWMutex
    checkInterval    time.Duration       // 检查间隔（如10秒）
    ctx              context.Context
    cancel           context.CancelFunc
}

// NewHealthChecker 创建健康检查器
func NewHealthChecker(checkInterval time.Duration) *HealthChecker {
    ctx, cancel := context.WithCancel(context.Background())
    return &HealthChecker{
        serviceInstances: make(map[string][]string),
        healthyInstances: make(map[string][]string),
        checkInterval:    checkInterval,
        ctx:              ctx,
        cancel:           cancel,
    }
}

// UpdateServiceInstances 更新服务实例列表（从服务发现获取）
func (hc *HealthChecker) UpdateServiceInstances(serviceName string, instances []string) {
    hc.mu.Lock()
    defer hc.mu.Unlock()
    hc.serviceInstances[serviceName] = instances
}

// Start 启动健康检查（定时执行）
func (hc *HealthChecker) Start() {
    ticker := time.NewTicker(hc.checkInterval)
    defer ticker.Stop()

    for range ticker.C {
        hc.checkAllServices()
    }
}

// checkAllServices 检查所有服务的实例健康状态
func (hc *HealthChecker) checkAllServices() {
    hc.mu.RLock()
    defer hc.mu.RUnlock()

    for serviceName, instances := range hc.serviceInstances {
        // 并发检查每个实例（提高效率）
        var wg sync.WaitGroup
        healthy := make([]string, 0, len(instances))
        for _, instance := range instances {
            wg.Add(1)
            go func(addr string) {
                defer wg.Done()
                if hc.checkInstance(addr) {
                    healthy = append(healthy, addr)
                }
            }(instance)
        }
        wg.Wait()

        // 更新健康实例列表
        hc.mu.Lock()
        hc.healthyInstances[serviceName] = healthy
        hc.mu.Unlock()

        fmt.Printf("服务[%s]健康检查结果：总实例数=%d，健康实例数=%d\n", serviceName, len(instances), len(healthy))
    }
}

// checkInstance 检查单个实例的健康状态（HTTP GET请求健康检查接口）
func (hc *HealthChecker) checkInstance(addr string) bool {
    // 健康检查接口：后端服务需提供（如"/health"）
    healthURL := addr + "/health"
    client := http.Client{
        Timeout: 3 * time.Second, // 超时时间3秒
    }

    resp, err := client.Get(healthURL)
    if err != nil || resp.StatusCode != http.StatusOK {
        return false
    }
    defer resp.Body.Close()
    return true
}

// GetHealthyInstances 获取服务的健康实例列表
func (hc *HealthChecker) GetHealthyInstances(serviceName string) []string {
    hc.mu.RLock()
    defer hc.mu.RUnlock()
    return hc.healthyInstances[serviceName]
}

// Close 关闭健康检查
func (hc *HealthChecker) Close() {
    hc.cancel()
}

package main

import (
    "sync"
    "math/rand"
    "time"
)

// CanaryManager 灰度发布管理器
type CanaryManager struct {
    serviceCanary map[string]*CanaryConfig // 服务名→灰度配置
    mu            sync.RWMutex
    random        *rand.Rand
}

// CanaryConfig 灰度配置
type CanaryConfig struct {
    Enable      bool     // 是否启用灰度
    NewService  string   // 新版本服务地址（如"http://localhost:8082"）
    OldService  string   // 旧版本服务地址（如"http://localhost:8081"）
    CanaryRatio int      // 灰度流量占比（0-100，如20表示20%流量走新版本）
}

// NewCanaryManager 创建灰度发布管理器
func NewCanaryManager() *CanaryManager {
    return &CanaryManager{
        serviceCanary: make(map[string]*CanaryConfig),
        random:        rand.New(rand.NewSource(time.Now().UnixNano())),
    }
}

// UpdateCanaryConfig 更新服务的灰度配置
func (cm *CanaryManager) UpdateCanaryConfig(serviceName string, config *CanaryConfig) {
    cm.mu.Lock()
    defer cm.mu.Unlock()
    cm.serviceCanary[serviceName] = config
}

// GetTargetService 根据灰度配置选择目标服务（旧版本/新版本）
func (cm *CanaryManager) GetTargetService(serviceName string) string {
    cm.mu.RLock()
    defer cm.mu.RUnlock()

    config, exists := cm.serviceCanary[serviceName]
    // 未启用灰度或配置不存在，走旧版本
    if !exists || !config.Enable {
        return config.OldService
    }

    // 基于权重选择：生成0-99的随机数，小于灰度占比则走新版本
    randomNum := cm.random.Intn(100)
    if randomNum < config.CanaryRatio {
        fmt.Printf("服务[%s]灰度发布：流量走新版本（占比：%d%%）\n", serviceName, config.CanaryRatio)
        return config.NewService
    }
    return config.OldService
}

// 集成到网关：在动态路由的MatchRoute中，调用GetTargetService选择目标服务
func integrateCanary() {
    // 1. 初始化灰度管理器
    cm := NewCanaryManager()

    // 2. 配置用户服务的灰度发布（20%流量走新版本）
    cm.UpdateCanaryConfig("user-service", &CanaryConfig{
        Enable:      true,
        OldService:  "http://localhost:8081", // 旧版本（v1）
        NewService:  "http://localhost:8082", // 新版本（v2）
        CanaryRatio: 20,                     // 20%流量
    })

    // 3. 在反向代理中，根据灰度结果选择目标服务
    // （修改reverseProxy函数，从cm.GetTargetService获取upstreamURL）
}