跳到主要内容

设计 API 网关

问题​

如何用 Python 设计一个 API 网关？包括路由、限流、认证等核心功能。

答案​

架构​

FastAPI 实现网关​

gateway/main.py
from fastapi import FastAPI, Request, HTTPException
import httpx

app = FastAPI(title="API Gateway")

# 路由表：前缀 → 后端服务
ROUTES = {
    "/api/users": "http://user-service:8001",
    "/api/orders": "http://order-service:8002",
    "/api/products": "http://product-service:8003",
}

@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def proxy(request: Request, path: str):
    target_url = resolve_target(path)
    if not target_url:
        raise HTTPException(status_code=404, detail="Route not found")

    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.request(
            method=request.method,
            url=target_url,
            headers=dict(request.headers),
            content=await request.body(),
        )
    return Response(content=response.content, status_code=response.status_code)


def resolve_target(path: str) -> str | None:
    for prefix, backend in ROUTES.items():
        if f"/{path}".startswith(prefix):
            return f"{backend}/{path}"
    return None

限流中间件​

gateway/middleware/rate_limit.py
import time
from collections import defaultdict
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware

class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, max_requests: int = 100, window: int = 60):
        super().__init__(app)
        self.max_requests = max_requests
        self.window = window
        # 生产环境用 Redis
        self.requests: dict[str, list[float]] = defaultdict(list)

    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host
        now = time.time()

        # 滑动窗口
        self.requests[client_ip] = [
            t for t in self.requests[client_ip] if t > now - self.window
        ]

        if len(self.requests[client_ip]) >= self.max_requests:
            raise HTTPException(status_code=429, detail="Rate limit exceeded")

        self.requests[client_ip].append(now)
        return await call_next(request)

JWT 认证中间件​

gateway/middleware/auth.py
import jwt
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware

# 不需要认证的白名单路径
PUBLIC_PATHS = {"/api/auth/login", "/api/auth/register", "/health"}

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if request.url.path in PUBLIC_PATHS:
            return await call_next(request)

        token = request.headers.get("Authorization", "").removeprefix("Bearer ")
        if not token:
            raise HTTPException(status_code=401, detail="Missing token")

        try:
            payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
            request.state.user_id = payload["sub"]
        except jwt.ExpiredSignatureError:
            raise HTTPException(status_code=401, detail="Token expired")
        except jwt.InvalidTokenError:
            raise HTTPException(status_code=401, detail="Invalid token")

        return await call_next(request)

熔断器​

gateway/circuit_breaker.py
from enum import Enum

class State(Enum):
    CLOSED = "closed"      # 正常，允许请求
    OPEN = "open"          # 熔断，拒绝请求
    HALF_OPEN = "half_open"  # 试探，放行部分请求

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 30):
        self.state = State.CLOSED
        self.failures = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = 0.0

    async def call(self, func, *args, **kwargs):
        if self.state == State.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = State.HALF_OPEN
            else:
                raise HTTPException(503, "Service unavailable (circuit open)")

        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        self.failures = 0
        self.state = State.CLOSED

    def _on_failure(self):
        self.failures += 1
        self.last_failure_time = time.time()
        if self.failures >= self.failure_threshold:
            self.state = State.OPEN

常见面试问题​

Q1: API 网关的核心功能有哪些？​

答案：

  1. 路由转发：根据路径前缀匹配后端服务
  2. 认证鉴权：统一验证 JWT/OAuth Token
  3. 限流：滑动窗口、令牌桶保护后端
  4. 熔断降级：后端异常时快速失败
  5. 协议转换：HTTP → gRPC、WebSocket 等
  6. 日志与追踪：统一记录请求日志、分配 TraceID

Q2: 令牌桶 vs 滑动窗口？​

答案：

算法特点适用场景
令牌桶允许突发、平滑限流API 限流
滑动窗口精确统计、无突发计数限流
漏桶固定速率、排队流量整形

Q3: Python vs Go/Java 做网关有什么优劣？​

答案：

  • 优势：开发效率高、FastAPI 异步性能不错、适合中小规模
  • 劣势：单进程 GIL、极高并发不如 Go/Java
  • 折中：用 Uvicorn 多 Worker + Nginx 前置可处理万级并发

相关链接​