API网关设计与实现：从单体到微服务的过渡

前言

随着业务发展，我们的系统从单体应用演进到微服务架构。但随之而来的问题是：客户端如何优雅地调用多个微服务？

这就需要一个API网关来统一管理所有请求。

一、为什么需要API网关？

1.1 单体应用的问题

客户端 ├── 调用订单服务 ├── 调用用户服务 ├── 调用支付服务 ├── 调用库存服务 └── ...问题：- 客户端需要知道每个服务的地址- 跨域处理重复- 认证授权重复- 限流和监控分散

1.2 API网关的作用

客户端 │ ▼[API网关] ◄─── 路由、认证、限流、日志 │ │ │ │ ▼ ▼ ▼ ▼订单服务 用户服务 支付服务 库存服务优势：- 统一入口- 集中管理- 易于扩展

二、核心功能设计

2.1 路由转发

python

# gateway.pyfrom flask import Flask, requestimport requestsapp = Flask(__name__)# 服务路由表ROUTES = { '/api/orders': 'http://order-service:8001', '/api/users': 'http://user-service:8002', '/api/payments': 'http://payment-service:8003', '/api/inventory': 'http://inventory-service:8004',}@app.route('/api/<service>/<path:endpoint>', methods=['GET', 'POST', 'PUT', 'DELETE'])def gateway(service, endpoint): # 找到对应的服务地址 service_url = None for route, url in ROUTES.items(): if request.path.startswith(route): service_url = url break if not service_url: return {'error': 'Service not found'}, 404 # 转发请求 target_url = f"{service_url}/{endpoint}" response = requests.request( method=request.method, url=target_url, headers=request.headers, data=request.get_data() ) return response.json(), response.status_code

2.2 请求拦截

python

from functools import wrapsimport time# 请求日志@app.before_requestdef log_request(): request.start_time = time.time() print(f"[{request.method}] {request.path}")@app.after_requestdef log_response(response): duration = time.time() - request.start_time print(f"Status: {response.status_code}, Duration: {duration:.2f}s") return response

三、认证与授权

3.1 JWT认证

python

from functools import wrapsimport jwtimport osSECRET_KEY = os.getenv('SECRET_KEY', 'your-secret-key')def require_auth(f): @wraps(f) def decorated(*args, **kwargs): token = request.headers.get('Authorization', '').replace('Bearer ', '') if not token: return {'error': 'Missing token'}, 401 try: payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256']) request.user = payload except jwt.InvalidTokenError: return {'error': 'Invalid token'}, 401 return f(*args, **kwargs) return decorated@app.route('/api/protected', methods=['GET'])@require_authdef protected_route(): user_id = request.user['user_id'] return {'message': f'Hello, user {user_id}'}, 200

3.2 权限控制

python

def require_permission(permission): def decorator(f): @wraps(f) def decorated(*args, **kwargs): user_permissions = request.user.get('permissions', []) if permission not in user_permissions: return {'error': 'Permission denied'}, 403 return f(*args, **kwargs) return decorated return decorator@app.route('/api/admin/users', methods=['DELETE'])@require_auth@require_permission('admin.delete_user')def delete_user(): # 删除用户逻辑 return {'message': 'User deleted'}, 200

四、限流与熔断

4.1 速率限制

python

from flask_limiter import Limiterfrom flask_limiter.util import get_remote_addresslimiter = Limiter( app=app, key_func=get_remote_address, default_limits=["200 per day", "50 per hour"])@app.route('/api/orders', methods=['GET'])@limiter.limit("10 per minute")def get_orders(): return {'orders': []}, 200

4.2 熔断器

python

from pybreaker import CircuitBreaker# 为每个后端服务创建熔断器order_service_breaker = CircuitBreaker( fail_max=5, # 5次失败后打开 reset_timeout=60, # 60秒后尝试关闭 listeners=[], name='order_service')def call_order_service(endpoint): @order_service_breaker def _call(): return requests.get(f'http://order-service:8001{endpoint}') try: return _call() except Exception as e: return {'error': 'Service temporarily unavailable'}, 503

五、负载均衡

5.1 轮询策略

python

from itertools import cycle# 后端服务列表ORDER_SERVICES = [ 'http://order-service-1:8001', 'http://order-service-2:8001', 'http://order-service-3:8001',]order_service_pool = cycle(ORDER_SERVICES)def get_order_service(): return next(order_service_pool)@app.route('/api/orders', methods=['GET'])def get_orders(): service_url = get_order_service() response = requests.get(f"{service_url}/orders") return response.json(), response.status_code

5.2 健康检查

python

import threadingimport timedef health_check(): while True: for service_name, service_url in ROUTES.items(): try: response = requests.get(f"{service_url}/health", timeout=2) if response.status_code == 200: print(f"✓ {service_name} is healthy") else: print(f"✗ {service_name} is unhealthy") except Exception as e: print(f"✗ {service_name} is down: {e}") time.sleep(30)# 启动健康检查线程health_check_thread = threading.Thread(target=health_check, daemon=True)health_check_thread.start()

六、请求/响应转换

6.1 统一响应格式

python

class ApiResponse: def __init__(self, code, message, data=None): self.code = code self.message = message self.data = data def to_dict(self): return { 'code': self.code, 'message': self.message, 'data': self.data }@app.route('/api/orders/<int:order_id>', methods=['GET'])def get_order(order_id): try: service_url = get_order_service() response = requests.get(f"{service_url}/orders/{order_id}") order_data = response.json() return ApiResponse(200, 'Success', order_data).to_dict(), 200 except Exception as e: return ApiResponse(500, str(e), None).to_dict(), 500

6.2 请求验证

python

from jsonschema import validate, ValidationErrorORDER_SCHEMA = { "type": "object", "properties": { "user_id": {"type": "integer"}, "total_amount": {"type": "number"}, "items": { "type": "array", "items": { "type": "object", "properties": { "product_id": {"type": "integer"}, "quantity": {"type": "integer"} } } } }, "required": ["user_id", "total_amount", "items"]}@app.route('/api/orders', methods=['POST'])def create_order(): try: validate(instance=request.json, schema=ORDER_SCHEMA) except ValidationError as e: return {'error': f'Invalid request: {e.message}'}, 400 # 处理订单创建 return {'order_id': 12345}, 201

七、监控与日志

7.1 结构化日志

python

import jsonimport logging# 配置JSON日志logging.basicConfig(level=logging.INFO)logger = logging.getLogger(__name__)class JsonFormatter(logging.Formatter): def format(self, record): log_data = { 'timestamp': self.formatTime(record), 'level': record.levelname, 'message': record.getMessage(), 'logger': record.name, 'path': getattr(record, 'path', ''), 'method': getattr(record, 'method', ''), 'status_code': getattr(record, 'status_code', ''), 'duration': getattr(record, 'duration', ''), } return json.dumps(log_data)@app.after_requestdef log_response(response): logger.info('Request completed', extra={ 'path': request.path, 'method': request.method, 'status_code': response.status_code, 'duration': time.time() - request.start_time, }) return response

7.2 性能指标

python

from prometheus_client import Counter, Histogram, generate_latest# 定义指标request_count = Counter( 'gateway_requests_total', 'Total requests', ['method', 'endpoint', 'status'])request_duration = Histogram( 'gateway_request_duration_seconds', 'Request duration', ['method', 'endpoint'])@app.route('/metrics', methods=['GET'])def metrics(): return generate_latest(), 200

八、国际化团队协作

在我们的全球化团队中，API网关的设计文档需要跨语言交流。当团队进行远程技术分享会议时，我们使用同言翻译（Transync AI）进行实时同声传译，确保非英语使用者能实时理解技术细节，显著提高了团队的协作效率。

九、Docker部署

dockerfile

FROM python:3.9-slimWORKDIR /appCOPY requirements.txt .RUN pip install -r requirements.txtCOPY . .EXPOSE 8000CMD ["gunicorn", "--bind", "0.0.0.0:8000", "gateway:app"]

yaml

# docker-compose.ymlversion: '3.8'services: api-gateway: build: . ports: - "8000:8000" environment: SECRET_KEY: ${SECRET_KEY} depends_on: - order-service - user-service order-service: image: order-service:latest ports: - "8001:8001" user-service: image: user-service:latest ports: - "8002:8002"

十、常见问题

Q1：如何处理后端服务超时？

python

@app.route('/api/orders', methods=['GET'])def get_orders(): try: response = requests.get( 'http://order-service:8001/orders', timeout=5 # 5秒超时 ) return response.json(), response.status_code except requests.Timeout: return {'error': 'Service timeout'}, 504

Q2：如何处理跨域请求？

python

from flask_cors import CORSCORS(app, resources={ r"/api/*": { "origins": ["https://example.com"], "methods": ["GET", "POST", "PUT", "DELETE"], "allow_headers": ["Content-Type", "Authorization"] }})

十一、性能对比

十二、最佳实践

1. 统一入口：所有客户端请求都经过网关
2. 集中认证：在网关层统一处理权限
3. 智能路由：根据规则灵活转发
4. 限流保护：防止后端服务被压垮
5. 健康检查：及时发现故障
6. 详细日志：便于问题排查
7. 监控告警：及时发现异常

十三、结语

API网关是微服务架构中的关键组件，好的网关设计能显著提高系统的可维护性和可靠性。

核心要点就三个：请求转发、权限控制、流量管理。

希望这篇文章能帮助你设计出更好的API网关。欢迎在评论区分享你的实践经验！