当前位置：首页 > news >正文

Go语言API网关设计与实现

news 2026/5/24 7:47:26

Go语言API网关设计与实现

引言

API网关是微服务架构的入口，负责统一管理所有外部请求。本文将深入探讨Go语言中API网关的设计与实现，包括路由、负载均衡、认证、限流等核心功能。

一、API网关概述

1.1 网关架构

外部请求 API网关 后端服务 | | | |--- 请求 ---> | | | |--- 认证/限流 -- | | | | | |--- 路由匹配 -- | | | | | |--- 负载均衡 -- | | | | | |<-- 响应 ------- | |<-- 响应 ------ | |

1.2 核心功能

功能	说明
请求路由	根据路径匹配后端服务
负载均衡	选择合适的服务实例
认证授权	验证用户身份和权限
限流熔断	保护后端服务
请求转发	代理请求到后端
日志监控	记录请求和指标

二、网关核心组件

2.1 路由管理器

type RouteManager struct { routes sync.Map logger *zap.Logger } type Route struct { ID string Path string Method string ServiceName string Middlewares []string Timeout time.Duration Headers map[string]string } func NewRouteManager() *RouteManager { return &RouteManager{ logger: zap.L().Named("route-manager"), } } func (rm *RouteManager) AddRoute(route Route) { key := fmt.Sprintf("%s:%s", route.Method, route.Path) rm.routes.Store(key, route) } func (rm *RouteManager) GetRoute(method, path string) (*Route, bool) { key := fmt.Sprintf("%s:%s", method, path) value, ok := rm.routes.Load(key) if !ok { return nil, false } route, _ := value.(Route) return &route, true } func (rm *RouteManager) RemoveRoute(method, path string) { key := fmt.Sprintf("%s:%s", method, path) rm.routes.Delete(key) }

2.2 URL路径匹配

type PathMatcher interface { Match(path string) (*Route, bool) } type ExactMatcher struct { routes map[string]*Route } func NewExactMatcher(routes []*Route) *ExactMatcher { m := &ExactMatcher{ routes: make(map[string]*Route), } for _, route := range routes { m.routes[route.Path] = route } return m } func (m *ExactMatcher) Match(path string) (*Route, bool) { route, ok := m.routes[path] return route, ok } type PatternMatcher struct { routes []*Route } func (m *PatternMatcher) Match(path string) (*Route, bool) { for _, route := range m.routes { if matched, _ := regexp.MatchString(route.Path, path); matched { return route, true } } return nil, false } type RadixTreeMatcher struct { root *radixNode } type radixNode struct { path string children map[string]*radixNode route *Route } func (m *RadixTreeMatcher) Insert(path string, route *Route) { m.root.insert(path, route) } func (n *radixNode) insert(path string, route *Route) { if path == "" { n.route = route return } for prefix, child := range n.children { if strings.HasPrefix(path, prefix) { child.insert(path[len(prefix):], route) return } } n.children[path] = &radixNode{ path: path, children: make(map[string]*radixNode), route: route, } } func (m *RadixTreeMatcher) Match(path string) (*Route, bool) { return m.root.match(path) } func (n *radixNode) match(path string) (*Route, bool) { if path == "" { return n.route, n.route != nil } for prefix, child := range n.children { if strings.HasPrefix(path, prefix) { return child.match(path[len(prefix):]) } } return nil, false }

三、中间件机制

3.1 中间件链

type Middleware func(http.Handler) http.Handler type MiddlewareChain struct { middlewares []Middleware } func NewMiddlewareChain(middlewares []Middleware) *MiddlewareChain { return &MiddlewareChain{ middlewares: middlewares, } } func (mc *MiddlewareChain) Then(handler http.Handler) http.Handler { for i := len(mc.middlewares) - 1; i >= 0; i-- { handler = mc.middlewares[i](handler) } return handler }

3.2 常用中间件

func LoggingMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { start := time.Now() logger := zap.L().Named("logging-middleware") rw := &responseWriter{ResponseWriter: w, statusCode: http.StatusOK} next.ServeHTTP(rw, r) duration := time.Since(start) logger.Info("Request completed", zap.String("method", r.Method), zap.String("path", r.URL.Path), zap.Int("status", rw.statusCode), zap.Duration("duration", duration), zap.String("remote_addr", r.RemoteAddr), ) }) } func AuthMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { token := r.Header.Get("Authorization") if token == "" { http.Error(w, "Unauthorized", http.StatusUnauthorized) return } claims, err := validateToken(token) if err != nil { http.Error(w, "Invalid token", http.StatusUnauthorized) return } ctx := context.WithValue(r.Context(), "user", claims) next.ServeHTTP(w, r.WithContext(ctx)) }) } func RateLimitMiddleware(limit int, window time.Duration) Middleware { limiter := NewTokenBucketLimiter(limit, window) return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if !limiter.Allow() { http.Error(w, "Too many requests", http.StatusTooManyRequests) return } next.ServeHTTP(w, r) }) } } func CORSHandler(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Access-Control-Allow-Origin", "*") w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization") if r.Method == http.MethodOptions { w.WriteHeader(http.StatusOK) return } next.ServeHTTP(w, r) }) }

四、请求代理

4.1 HTTP代理

type Proxy struct { httpClient *http.Client logger *zap.Logger } func NewProxy(timeout time.Duration) *Proxy { return &Proxy{ httpClient: &http.Client{ Timeout: timeout, Transport: &http.Transport{ MaxIdleConns: 100, IdleConnTimeout: 30 * time.Second, TLSHandshakeTimeout: 10 * time.Second, }, }, logger: zap.L().Named("proxy"), } } func (p *Proxy) Proxy(w http.ResponseWriter, r *http.Request, targetURL string) error { proxyReq, err := http.NewRequest(r.Method, targetURL, r.Body) if err != nil { return err } for key, values := range r.Header { for _, value := range values { proxyReq.Header.Add(key, value) } } resp, err := p.httpClient.Do(proxyReq) if err != nil { return err } defer resp.Body.Close() for key, values := range resp.Header { for _, value := range values { w.Header().Add(key, value) } } w.WriteHeader(resp.StatusCode) _, err = io.Copy(w, resp.Body) return err }

4.2 gRPC代理

type gRPCProxy struct { logger *zap.Logger } func (p *gRPCProxy) Handle(w http.ResponseWriter, r *http.Request, backendAddr string) { conn, err := grpc.Dial(backendAddr, grpc.WithInsecure()) if err != nil { http.Error(w, "Failed to connect to backend", http.StatusBadGateway) return } defer conn.Close() grpcHandler := func(ctx context.Context, req *http.Request) (*http.Response, error) { backendReq, err := http.NewRequest(req.Method, "http://"+backendAddr+req.URL.Path, req.Body) if err != nil { return nil, err } return p.httpClient.Do(backendReq) } grpc.ServeHTTP(w, r, grpc.HandlerFunc(grpcHandler)) }

五、负载均衡

5.1 负载均衡接口

type LoadBalancer interface { Next(instances []*ServiceInstance) *ServiceInstance } type ServiceInstance struct { ID string Name string Address string Port int Weight int }

5.2 轮询算法

type RoundRobinBalancer struct { mu sync.Mutex current int } func (b *RoundRobinBalancer) Next(instances []*ServiceInstance) *ServiceInstance { if len(instances) == 0 { return nil } b.mu.Lock() defer b.mu.Unlock() instance := instances[b.current] b.current = (b.current + 1) % len(instances) return instance }

5.3 最少连接数算法

type LeastConnectionsBalancer struct { mu sync.Mutex instances []*ServiceInstance counters map[string]int } func (b *LeastConnectionsBalancer) Next(instances []*ServiceInstance) *ServiceInstance { if len(instances) == 0 { return nil } b.mu.Lock() defer b.mu.Unlock() // 更新实例列表 for _, instance := range instances { if _, exists := b.counters[instance.ID]; !exists { b.counters[instance.ID] = 0 } } // 找到连接数最少的实例 minConn := b.counters[instances[0].ID] selected := instances[0] for _, instance := range instances[1:] { if b.counters[instance.ID] < minConn { minConn = b.counters[instance.ID] selected = instance } } b.counters[selected.ID]++ return selected } func (b *LeastConnectionsBalancer) Release(instanceID string) { b.mu.Lock() defer b.mu.Unlock() if count, exists := b.counters[instanceID]; exists && count > 0 { b.counters[instanceID]-- } }

5.4 IP哈希算法

type IPHashBalancer struct{} func (b *IPHashBalancer) Next(instances []*ServiceInstance, clientIP string) *ServiceInstance { if len(instances) == 0 { return nil } hash := crc32.ChecksumIEEE([]byte(clientIP)) index := int(hash) % len(instances) return instances[index] }

六、限流与熔断

6.1 令牌桶限流

type TokenBucketLimiter struct { mu sync.Mutex capacity int tokens int refillRate time.Duration lastRefill time.Time } func NewTokenBucketLimiter(capacity int, refillRate time.Duration) *TokenBucketLimiter { return &TokenBucketLimiter{ capacity: capacity, tokens: capacity, refillRate: refillRate, lastRefill: time.Now(), } } func (l *TokenBucketLimiter) Allow() bool { l.mu.Lock() defer l.mu.Unlock() now := time.Now() elapsed := now.Sub(l.lastRefill) // 计算应该补充的令牌数 tokensToAdd := int(elapsed / l.refillRate) if tokensToAdd > 0 { l.tokens = min(l.tokens+tokensToAdd, l.capacity) l.lastRefill = now } if l.tokens > 0 { l.tokens-- return true } return false } func min(a, b int) int { if a < b { return a } return b }

6.2 熔断器

type CircuitBreaker struct { mu sync.Mutex state CircuitState failureThreshold int successThreshold int failureCount int successCount int timeout time.Duration lastFailure time.Time } type CircuitState int const ( Closed CircuitState = iota Open HalfOpen ) func NewCircuitBreaker(failureThreshold, successThreshold int, timeout time.Duration) *CircuitBreaker { return &CircuitBreaker{ state: Closed, failureThreshold: failureThreshold, successThreshold: successThreshold, timeout: timeout, } } func (cb *CircuitBreaker) Execute(fn func() error) error { cb.mu.Lock() state := cb.state cb.mu.Unlock() switch state { case Open: if time.Since(cb.lastFailure) > cb.timeout { cb.mu.Lock() cb.state = HalfOpen cb.mu.Unlock() } else { return fmt.Errorf("circuit breaker is open") } } err := fn() cb.mu.Lock() defer cb.mu.Unlock() if err != nil { cb.failureCount++ cb.successCount = 0 cb.lastFailure = time.Now() if cb.state == HalfOpen && cb.failureCount >= cb.failureThreshold { cb.state = Open } } else { cb.successCount++ cb.failureCount = 0 if cb.state == HalfOpen && cb.successCount >= cb.successThreshold { cb.state = Closed } } return err }

七、监控与日志

7.1 指标收集

type MetricsCollector struct { requestCount *prometheus.CounterVec requestDuration *prometheus.HistogramVec errorCount *prometheus.CounterVec activeConnections prometheus.Gauge } func NewMetricsCollector(registry *prometheus.Registry) *MetricsCollector { mc := &MetricsCollector{ requestCount: prometheus.NewCounterVec( prometheus.CounterOpts{ Name: "gateway_requests_total", Help: "Total number of requests", }, []string{"method", "path", "status"}, ), requestDuration: prometheus.NewHistogramVec( prometheus.HistogramOpts{ Name: "gateway_request_duration_seconds", Help: "Request duration in seconds", Buckets: prometheus.DefBuckets, }, []string{"method", "path"}, ), errorCount: prometheus.NewCounterVec( prometheus.CounterOpts{ Name: "gateway_errors_total", Help: "Total number of errors", }, []string{"method", "path", "error"}, ), activeConnections: prometheus.NewGauge( prometheus.GaugeOpts{ Name: "gateway_active_connections", Help: "Number of active connections", }, ), } registry.MustRegister( mc.requestCount, mc.requestDuration, mc.errorCount, mc.activeConnections, ) return mc } func (mc *MetricsCollector) RecordRequest(method, path string, status int, duration time.Duration) { mc.requestCount.WithLabelValues(method, path, fmt.Sprintf("%d", status)).Inc() mc.requestDuration.WithLabelValues(method, path).Observe(duration.Seconds()) } func (mc *MetricsCollector) RecordError(method, path, err string) { mc.errorCount.WithLabelValues(method, path, err).Inc() }

7.2 结构化日志

func initLogger() *zap.Logger { cfg := zap.Config{ Level: zap.NewAtomicLevelAt(zap.InfoLevel), Development: false, Sampling: &zap.SamplingConfig{ Initial: 100, Thereafter: 100, }, Encoding: "json", EncoderConfig: zap.NewProductionEncoderConfig(), OutputPaths: []string{"stdout"}, ErrorOutputPaths: []string{"stderr"}, } logger, _ := cfg.Build() return logger } func logRequest(logger *zap.Logger, r *http.Request, status int, duration time.Duration) { logger.Info("request", zap.String("method", r.Method), zap.String("path", r.URL.Path), zap.String("query", r.URL.RawQuery), zap.String("remote_addr", r.RemoteAddr), zap.Int("status", status), zap.Duration("duration", duration), zap.String("user_agent", r.UserAgent()), ) }

八、配置管理

8.1 动态配置

type GatewayConfig struct { Port int Timeout time.Duration Routes []RouteConfig Middlewares []MiddlewareConfig RateLimit RateLimitConfig CircuitBreaker CircuitBreakerConfig } type RouteConfig struct { ID string Path string Method string ServiceName string Timeout time.Duration } func LoadConfig(path string) (*GatewayConfig, error) { data, err := os.ReadFile(path) if err != nil { return nil, err } var config GatewayConfig if err := yaml.Unmarshal(data, &config); err != nil { return nil, err } return &config, nil } func WatchConfig(path string, callback func(*GatewayConfig)) { fsWatcher, err := fs.NewWatcher() if err != nil { return } go func() { for { select { case event := <-fsWatcher.Events: if event.Has(fs.Write) { config, err := LoadConfig(path) if err != nil { continue } callback(config) } case err := <-fsWatcher.Errors: log.Printf("Watcher error: %v", err) } } }() fsWatcher.Add(path) }

九、完整网关实现

type Gateway struct { router *http.ServeMux routeManager *RouteManager proxy *Proxy loadBalancer LoadBalancer middlewareChain *MiddlewareChain metrics *MetricsCollector logger *zap.Logger } func NewGateway(config *GatewayConfig) *Gateway { gw := &Gateway{ router: http.NewServeMux(), routeManager: NewRouteManager(), proxy: NewProxy(config.Timeout), loadBalancer: &RoundRobinBalancer{}, logger: initLogger(), } // 注册路由 for _, routeConfig := range config.Routes { route := Route{ ID: routeConfig.ID, Path: routeConfig.Path, Method: routeConfig.Method, ServiceName: routeConfig.ServiceName, Timeout: routeConfig.Timeout, } gw.routeManager.AddRoute(route) } // 构建中间件链 middlewares := []Middleware{ LoggingMiddleware, CORSHandler, RateLimitMiddleware(config.RateLimit.Limit, config.RateLimit.Window), } gw.middlewareChain = NewMiddlewareChain(middlewares) // 设置处理函数 gw.router.HandleFunc("/", gw.handleRequest) return gw } func (gw *Gateway) handleRequest(w http.ResponseWriter, r *http.Request) { start := time.Now() // 查找路由 route, ok := gw.routeManager.GetRoute(r.Method, r.URL.Path) if !ok { http.Error(w, "Route not found", http.StatusNotFound) gw.logger.Error("Route not found", zap.String("path", r.URL.Path)) return } // 服务发现 instances, err := gw.discoverService(route.ServiceName) if err != nil || len(instances) == 0 { http.Error(w, "Service unavailable", http.StatusServiceUnavailable) gw.logger.Error("Service unavailable", zap.String("service", route.ServiceName)) return } // 负载均衡 instance := gw.loadBalancer.Next(instances) targetURL := fmt.Sprintf("http://%s:%d%s", instance.Address, instance.Port, r.URL.Path) // 请求代理 if err := gw.proxy.Proxy(w, r, targetURL); err != nil { http.Error(w, "Proxy error", http.StatusBadGateway) gw.logger.Error("Proxy error", zap.Error(err)) } duration := time.Since(start) gw.logger.Info("Request completed", zap.String("method", r.Method), zap.String("path", r.URL.Path), zap.String("service", route.ServiceName), zap.Duration("duration", duration), ) } func (gw *Gateway) Serve(port int) error { addr := fmt.Sprintf(":%d", port) gw.logger.Info("Gateway starting", zap.String("addr", addr)) server := &http.Server{ Addr: addr, Handler: gw.middlewareChain.Then(gw.router), } return server.ListenAndServe() }