第39章：前后端分离与API网关

🌟 章节导入：走进现代化企业服务架构中心

亲爱的朋友们，欢迎来到我们的现代化企业服务架构中心！这是一个充满创新活力的智能化办公园区，在这里，我们将见证现代Web应用如何通过前后端分离和API网关技术，实现高效、安全、可扩展的企业级服务架构。

🏢 企业服务架构中心全景

想象一下，你正站在一个现代化的科技园区门口，眼前是四座风格迥异但又紧密相连的建筑群：

🏗️ 前后端分离架构大厦

这是我们的第一站，一座现代化的前后端分离办公大厦。在这里：

• 前端开发部里，工程师们正在构建现代化的用户界面，使用React、Vue等框架
• 后端服务部的专家们专注于业务逻辑和数据处理，提供标准化的API服务
• 通信协调部如同专业的信使团队，确保前后端之间的数据交互顺畅高效

🚪 API网关服务中心

这座建筑闪烁着蓝色的光芒，象征着智能化的服务调度中心：

• 路由调度室里，智能系统根据请求特征将流量分发到不同的后端服务
• 限流保护系统如同一道坚固的防护墙，防止系统过载和恶意攻击
• 服务发现中心汇聚了所有微服务的信息，实现动态的服务注册与发现

🛡️ 安全防护中心

这是一座充满安全感的网络安全防护堡垒：

• 跨域处理室如同专业的边界检查站，确保跨域请求的安全传输
• 认证授权部里，JWT令牌系统如同智能门禁，验证每一个访问请求
• 安全审计中心是整个安全体系的核心，记录和分析所有安全事件

🌐 微服务API网关平台

最令人兴奋的是这座未来感十足的智能服务调度平台：

• 统一认证中心如同企业总部的身份验证系统，管理所有服务的访问权限
• 服务监控大屏展示着实时系统状态，帮助运维团队快速发现问题
• 流量控制中心是整个平台的大脑，负责智能化的流量分配和负载均衡

🚀 技术革命的见证者

在这个企业服务架构中心，我们将见证Web应用架构的三大革命：

🏗️ 架构分离革命

从传统的单体应用到前后端分离架构，我们将掌握：

• 前后端职责的清晰划分
• 独立开发和部署的能力
• 技术栈选择的灵活性

🚪 网关服务革命

从直接访问后端服务到通过API网关统一管理，我们将实现：

• 统一的入口和路由管理
• 智能的流量控制和负载均衡
• 完善的监控和安全防护

🛡️ 安全防护革命

从基础的安全措施到企业级的安全体系，我们将建立：

• 完善的跨域和安全处理机制
• 标准化的认证授权流程
• 全面的安全审计和监控

🎯 学以致用的企业级项目

在本章的最后，我们将综合运用所学的所有技术，构建一个完整的微服务API网关系统。这不仅仅是一个学习项目，更是一个具备实际商业部署价值的企业级应用：

• 企业应用可以集成这个系统，实现统一的API管理和安全控制
• 微服务架构可以基于这个系统，实现服务的统一入口和智能调度
• 云原生应用可以利用这个系统，实现高可用和自动扩展的服务架构
• 技术服务商可以基于这个系统为客户提供定制化的API网关解决方案

🔥 准备好了吗？

现在，让我们戴上安全帽，穿上工作服，一起走进这个充满科技魅力的企业服务架构中心。在这里，我们不仅要学习最前沿的Web架构技术，更要将这些技术转化为真正有价值的商业应用！

准备好迎接这场技术革命了吗？让我们开始这激动人心的学习之旅！

---

🎯 学习目标（SMART目标）

完成本章学习后，学生将能够：

📚 知识目标

• 前后端分离架构体系：深入理解前后端分离的设计原则、数据交互规范、状态管理策略等核心概念
• API网关技术：掌握API网关的架构设计、路由和负载均衡、限流和熔断机制、服务发现与注册等关键技术
• 跨域和安全技术：理解CORS配置管理、JWT令牌机制、API安全防护等安全技术
• 微服务架构理念：综合运用服务拆分、服务治理、统一认证等微服务架构技术

🛠️ 技能目标

• 架构设计能力：能够独立设计前后端分离的应用架构，实现前后端的独立开发和部署
• API网关开发能力：具备API网关的设计和实现能力，包括路由、限流、监控等功能
• 安全防护能力：掌握跨域处理、JWT认证、API安全防护的实战能力
• 企业级部署能力：能够构建完整的微服务API网关系统，具备大规模部署的工程实践能力

💡 素养目标

• 架构设计思维：培养现代化应用架构设计的前瞻性思维模式
• 安全防护意识：建立完善的安全防护和合规性意识
• 服务治理理念：注重高可用、可扩展、可维护的服务治理理念
• 工程实践能力：理解企业级应用开发的核心要求和最佳实践

---

📝 知识导图

---

🎓 理论讲解

39.1 前后端分离架构

想象一下，您走进了一家现代化的企业总部。首先映入眼帘的是前后端分离架构大厦——这座大厦分为两个独立的办公区域：前端开发部和后端服务部。它们各自专注于自己的核心职责，通过标准化的通信协议进行协作，就像两个专业团队通过电话和邮件进行高效沟通一样。

在Web应用开发的世界里，前后端分离架构就是我们的"现代化办公模式"。它将用户界面（前端）和业务逻辑（后端）完全分离，让两个团队可以独立开发、测试和部署，大大提高了开发效率和系统的可维护性。

🔧 前后端分离的核心原理

传统架构 vs 前后端分离架构

让我们用企业办公的例子来理解这两种架构的差异：

# 示例1：理解前后端分离架构的核心优势

"""

前后端分离架构演示：现代化Web应用架构的标准化解决方案



比喻说明：

- 传统架构 = 一个部门负责所有工作（前端+后端+数据库）

- 前后端分离 = 前端部门+后端部门，各司其职，通过API协作

"""



import json

import time

from datetime import datetime

from typing import Dict, List, Any

from dataclasses import dataclass, asdict



@dataclass

class APIRequest:

    """API请求数据模型"""

    method: str

    endpoint: str

    headers: Dict[str, str]

    body: Dict[str, Any]

    timestamp: str



@dataclass

class APIResponse:

    """API响应数据模型"""

    status_code: int

    data: Dict[str, Any]

    message: str

    timestamp: str



class FrontendBackendSeparationDemo:

    """前后端分离架构演示中心"""

    

    def __init__(self):

        """初始化演示环境"""

        self.api_requests = []

        self.api_responses = []

        print("🚀 前后端分离架构演示中心启动成功！")

    

    def compare_architectures(self):

        """对比传统架构与前后端分离架构"""

        print("\n" + "="*60)

        print("📊 传统架构 vs 前后端分离架构对比分析")

        print("="*60)

        

        # 传统架构特点

        traditional_arch = {

            "耦合度": "高（前后端紧密耦合）",

            "开发效率": "低（需要协调前后端）",

            "技术栈": "受限（必须使用相同技术）",

            "部署方式": "一体化部署",

            "可扩展性": "差（难以独立扩展）",

            "团队协作": "需要紧密配合"

        }

        

        # 前后端分离架构特点

        separated_arch = {

            "耦合度": "低（通过API解耦）",

            "开发效率": "高（可并行开发）",

            "技术栈": "灵活（前后端可独立选择）",

            "部署方式": "独立部署",

            "可扩展性": "好（可独立扩展）",

            "团队协作": "通过API契约协作"

        }

        

        print("🏢 传统架构（单体应用）：")

        for key, value in traditional_arch.items():

            print(f"   {key}: {value}")

            

        print("\n🌐 前后端分离架构（现代化方案）：")

        for key, value in separated_arch.items():

            print(f"   {key}: {value}")

            

        print("\n💡 结论：前后端分离架构在灵活性和可维护性上有显著优势！")

    

    def demonstrate_api_contract(self):

        """演示API契约设计"""

        print("\n" + "="*60)

        print("📝 API契约设计演示")

        print("="*60)

        

        # API契约示例

        api_contract = {

            "用户管理API": {

                "GET /api/users": {

                    "description": "获取用户列表",

                    "parameters": {

                        "page": "int, 页码",

                        "page_size": "int, 每页数量",

                        "search": "string, 搜索关键词"

                    },

                    "response": {

                        "status_code": 200,

                        "data": {

                            "users": "List[User], 用户列表",

                            "total": "int, 总数量",

                            "page": "int, 当前页码"

                        }

                    }

                },

                "POST /api/users": {

                    "description": "创建新用户",

                    "request_body": {

                        "username": "string, 用户名",

                        "email": "string, 邮箱",

                        "password": "string, 密码"

                    },

                    "response": {

                        "status_code": 201,

                        "data": {

                            "user": "User, 创建的用户对象",

                            "message": "string, 成功消息"

                        }

                    }

                },

                "GET /api/users/{id}": {

                    "description": "获取指定用户",

                    "path_parameters": {

                        "id": "int, 用户ID"

                    },

                    "response": {

                        "status_code": 200,

                        "data": {

                            "user": "User, 用户对象"

                        }

                    }

                }

            }

        }

        

        print("📋 API契约设计示例：")

        print(json.dumps(api_contract, indent=2, ensure_ascii=False))

        

        return api_contract

    

    def simulate_frontend_request(self, endpoint: str, method: str = "GET", data: Dict = None):

        """模拟前端请求后端API"""

        print(f"\n📤 前端发送请求：{method} {endpoint}")

        

        # 创建请求对象

        request = APIRequest(

            method=method,

            endpoint=endpoint,

            headers={

                "Content-Type": "application/json",

                "Authorization": "Bearer token_12345"

            },

            body=data or {},

            timestamp=datetime.now().isoformat()

        )

        

        self.api_requests.append(request)

        

        # 模拟后端处理

        response = self._process_backend_request(request)

        self.api_responses.append(response)

        

        print(f"📥 后端返回响应：状态码 {response.status_code}")

        print(f"   数据：{json.dumps(response.data, indent=2, ensure_ascii=False)}")

        

        return response

    

    def _process_backend_request(self, request: APIRequest) -> APIResponse:

        """模拟后端处理请求"""

        # 模拟处理延迟

        time.sleep(0.1)

        

        if request.method == "GET":

            if request.endpoint == "/api/users":

                # 模拟获取用户列表

                return APIResponse(

                    status_code=200,

                    data={

                        "users": [

                            {"id": 1, "username": "alice", "email": "alice@example.com"},

                            {"id": 2, "username": "bob", "email": "bob@example.com"}

                        ],

                        "total": 2,

                        "page": 1

                    },

                    message="获取用户列表成功",

                    timestamp=datetime.now().isoformat()

                )

            elif request.endpoint.startswith("/api/users/"):

                # 模拟获取单个用户

                user_id = request.endpoint.split("/")[-1]

                return APIResponse(

                    status_code=200,

                    data={

                        "user": {

                            "id": int(user_id),

                            "username": f"user_{user_id}",

                            "email": f"user_{user_id}@example.com"

                        }

                    },

                    message="获取用户成功",

                    timestamp=datetime.now().isoformat()

                )

        

        elif request.method == "POST":

            if request.endpoint == "/api/users":

                # 模拟创建用户

                return APIResponse(

                    status_code=201,

                    data={

                        "user": {

                            "id": 3,

                            "username": request.body.get("username", "new_user"),

                            "email": request.body.get("email", "new@example.com")

                        },

                        "message": "用户创建成功"

                    },

                    message="用户创建成功",

                    timestamp=datetime.now().isoformat()

                )

        

        # 默认响应

        return APIResponse(

            status_code=404,

            data={},

            message="接口不存在",

            timestamp=datetime.now().isoformat()

        )

    

    def demonstrate_state_management(self):

        """演示前后端状态管理策略"""

        print("\n" + "="*60)

        print("🔄 前后端状态管理策略演示")

        print("="*60)

        

        state_management = {

            "前端状态管理": {

                "客户端状态": "用户界面状态、表单数据、UI交互状态",

                "本地存储": "localStorage、sessionStorage、IndexedDB",

                "状态管理库": "Redux、Vuex、MobX等",

                "特点": "快速响应、减少服务器请求"

            },

            "后端状态管理": {

                "服务端状态": "用户会话、业务数据、系统配置",

                "数据库存储": "持久化数据、关系数据",

                "缓存系统": "Redis、Memcached等",

                "特点": "数据一致性、安全性"

            },

            "Token机制": {

                "JWT Token": "无状态认证、可跨域使用",

                "Refresh Token": "安全刷新、延长会话",

                "Token存储": "前端存储、HttpOnly Cookie",

                "特点": "无状态、可扩展"

            }

        }

        

        print("📋 状态管理策略：")

        print(json.dumps(state_management, indent=2, ensure_ascii=False))

        

        return state_management



# 运行演示

if __name__ == "__main__":

    demo = FrontendBackendSeparationDemo()

    

    # 对比架构

    demo.compare_architectures()

    

    # 演示API契约

    demo.demonstrate_api_contract()

    

    # 模拟前端请求

    demo.simulate_frontend_request("/api/users", "GET")

    demo.simulate_frontend_request("/api/users/1", "GET")

    demo.simulate_frontend_request("/api/users", "POST", {

        "username": "charlie",

        "email": "charlie@example.com",

        "password": "secure_password"

    })

    

    # 演示状态管理

    demo.demonstrate_state_management()

运行结果：

🚀 前后端分离架构演示中心启动成功！

============================================================
📊 传统架构 vs 前后端分离架构对比分析
============================================================
🏢 传统架构（单体应用）：
   耦合度: 高（前后端紧密耦合）
   开发效率: 低（需要协调前后端）
   技术栈: 受限（必须使用相同技术）
   部署方式: 一体化部署
   可扩展性: 差（难以独立扩展）
   团队协作: 需要紧密配合

🌐 前后端分离架构（现代化方案）：
   耦合度: 低（通过API解耦）
   开发效率: 高（可并行开发）
   技术栈: 灵活（前后端可独立选择）
   部署方式: 独立部署
   可扩展性: 好（可独立扩展）
   团队协作: 通过API契约协作

💡 结论：前后端分离架构在灵活性和可维护性上有显著优势！

前后端分离架构设计原则

在我们的"前后端分离架构大厦"中，有一套经过验证的设计原则：

# 示例2：前后端分离架构设计原则系统

"""

前后端分离架构设计原则的企业级最佳实践



包含：

- 职责分离原则

- 接口设计规范

- 数据交互标准

- 错误处理机制

"""



from abc import ABC, abstractmethod

from typing import Dict, Any, Optional

from enum import Enum

import json



class HTTPMethod(Enum):

    """HTTP方法枚举"""

    GET = "GET"

    POST = "POST"

    PUT = "PUT"

    DELETE = "DELETE"

    PATCH = "PATCH"



class APIResponseStatus(Enum):

    """API响应状态枚举"""

    SUCCESS = 200

    CREATED = 201

    BAD_REQUEST = 400

    UNAUTHORIZED = 401

    FORBIDDEN = 403

    NOT_FOUND = 404

    INTERNAL_ERROR = 500



class SeparationArchitecturePrinciples:

    """前后端分离架构设计原则系统"""

    

    def __init__(self):

        """初始化原则系统"""

        self.principles = {

            "职责分离": self._principle_separation_of_concerns,

            "接口规范": self._principle_api_standards,

            "数据格式": self._principle_data_format,

            "错误处理": self._principle_error_handling,

            "安全设计": self._principle_security_design

        }

        print("🏗️ 前后端分离架构设计原则系统启动")

    

    def _principle_separation_of_concerns(self):

        """职责分离原则"""

        return {

            "前端职责": [

                "用户界面渲染",

                "用户交互处理",

                "客户端状态管理",

                "数据展示和格式化"

            ],

            "后端职责": [

                "业务逻辑处理",

                "数据持久化",

                "安全认证授权",

                "API接口提供"

            ],

            "通信方式": [

                "RESTful API",

                "GraphQL",

                "WebSocket（实时通信）"

            ]

        }

    

    def _principle_api_standards(self):

        """接口规范原则"""

        return {

            "RESTful设计": {

                "资源导向": "URL表示资源，如 /api/users",

                "HTTP方法": "GET/POST/PUT/DELETE表示操作",

                "状态码": "使用标准HTTP状态码",

                "版本控制": "API版本化，如 /api/v1/users"

            },

            "命名规范": {

                "URL命名": "使用名词，复数形式，小写+连字符",

                "参数命名": "驼峰命名或下划线命名，保持一致",

                "响应格式": "统一的JSON格式"

            }

        }

    

    def _principle_data_format(self):

        """数据格式原则"""

        return {

            "请求格式": {

                "Content-Type": "application/json",

                "字符编码": "UTF-8",

                "日期格式": "ISO 8601格式"

            },

            "响应格式": {

                "成功响应": {

                    "status": "success",

                    "data": "实际数据",

                    "message": "可选消息"

                },

                "错误响应": {

                    "status": "error",

                    "error_code": "错误代码",

                    "error_message": "错误消息",

                    "details": "错误详情（可选）"

                }

            }

        }

    

    def _principle_error_handling(self):

        """错误处理原则"""

        return {

            "错误分类": {

                "客户端错误": "4xx状态码，如400、401、403、404",

                "服务器错误": "5xx状态码，如500、502、503",

                "业务错误": "使用自定义错误码和消息"

            },

            "错误响应格式": {

                "status_code": "HTTP状态码",

                "error": {

                    "code": "业务错误码",

                    "message": "用户友好的错误消息",

                    "details": "开发调试信息（可选）"

                }

            }

        }

    

    def _principle_security_design(self):

        """安全设计原则"""

        return {

            "认证机制": [

                "JWT Token认证",

                "OAuth2授权",

                "API Key认证"

            ],

            "安全措施": [

                "HTTPS传输加密",

                "输入验证和过滤",

                "SQL注入防护",

                "XSS攻击防护",

                "CSRF防护"

            ],

            "权限控制": [

                "基于角色的访问控制（RBAC）",

                "基于资源的权限控制",

                "API访问频率限制"

            ]

        }

    

    def get_all_principles(self):

        """获取所有设计原则"""

        result = {}

        for name, method in self.principles.items():

            result[name] = method()

        return result

    

    def print_principles(self):

        """打印所有设计原则"""

        print("\n" + "="*60)

        print("📋 前后端分离架构设计原则")

        print("="*60)

        

        principles = self.get_all_principles()

        print(json.dumps(principles, indent=2, ensure_ascii=False))



# 运行演示

if __name__ == "__main__":

    principles = SeparationArchitecturePrinciples()

    principles.print_principles()

Django REST Framework实现前后端分离

现在让我们看看如何在Django中实现前后端分离架构：

# 示例3：Django REST Framework前后端分离实现

"""

使用Django REST Framework实现前后端分离架构



包含：

- 模型定义

- 序列化器设计

- 视图集实现

- 路由配置

"""



# models.py - 数据模型定义

from django.db import models

from django.contrib.auth.models import AbstractUser



class User(AbstractUser):

    """用户模型"""

    phone = models.CharField(max_length=20, blank=True, null=True)

    avatar = models.URLField(blank=True, null=True)

    created_at = models.DateTimeField(auto_now_add=True)

    updated_at = models.DateTimeField(auto_now=True)

    

    class Meta:

        db_table = 'users'

        verbose_name = '用户'

        verbose_name_plural = '用户'



class Product(models.Model):

    """产品模型"""

    name = models.CharField(max_length=200, verbose_name='产品名称')

    description = models.TextField(blank=True, verbose_name='产品描述')

    price = models.DecimalField(max_digits=10, decimal_places=2, verbose_name='价格')

    stock = models.IntegerField(default=0, verbose_name='库存')

    category = models.ForeignKey('Category', on_delete=models.CASCADE, verbose_name='分类')

    created_at = models.DateTimeField(auto_now_add=True)

    updated_at = models.DateTimeField(auto_now=True)

    

    class Meta:

        db_table = 'products'

        verbose_name = '产品'

        verbose_name_plural = '产品'

        ordering = ['-created_at']



class Category(models.Model):

    """分类模型"""

    name = models.CharField(max_length=100, unique=True, verbose_name='分类名称')

    description = models.TextField(blank=True, verbose_name='分类描述')

    created_at = models.DateTimeField(auto_now_add=True)

    

    class Meta:

        db_table = 'categories'

        verbose_name = '分类'

        verbose_name_plural = '分类'



# serializers.py - 序列化器定义

from rest_framework import serializers

from .models import User, Product, Category



class CategorySerializer(serializers.ModelSerializer):

    """分类序列化器"""

    class Meta:

        model = Category

        fields = ['id', 'name', 'description', 'created_at']



class ProductSerializer(serializers.ModelSerializer):

    """产品序列化器"""

    category = CategorySerializer(read_only=True)

    category_id = serializers.IntegerField(write_only=True)

    

    class Meta:

        model = Product

        fields = ['id', 'name', 'description', 'price', 'stock', 

                 'category', 'category_id', 'created_at', 'updated_at']

        read_only_fields = ['id', 'created_at', 'updated_at']

    

    def validate_price(self, value):

        """价格验证"""

        if value <= 0:

            raise serializers.ValidationError("价格必须大于0")

        return value

    

    def validate_stock(self, value):

        """库存验证"""

        if value < 0:

            raise serializers.ValidationError("库存不能为负数")

        return value



class UserSerializer(serializers.ModelSerializer):

    """用户序列化器"""

    class Meta:

        model = User

        fields = ['id', 'username', 'email', 'phone', 'avatar', 

                 'created_at', 'updated_at']

        read_only_fields = ['id', 'created_at', 'updated_at']



# views.py - 视图集实现

from rest_framework import viewsets, status

from rest_framework.decorators import action

from rest_framework.response import Response

from rest_framework.permissions import IsAuthenticated, IsAuthenticatedOrReadOnly

from django_filters.rest_framework import DjangoFilterBackend

from rest_framework.filters import SearchFilter, OrderingFilter

from .models import Product, Category, User

from .serializers import ProductSerializer, CategorySerializer, UserSerializer



class ProductViewSet(viewsets.ModelViewSet):

    """产品视图集"""

    queryset = Product.objects.all()

    serializer_class = ProductSerializer

    permission_classes = [IsAuthenticatedOrReadOnly]

    filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]

    filterset_fields = ['category', 'stock']

    search_fields = ['name', 'description']

    ordering_fields = ['price', 'created_at', 'stock']

    ordering = ['-created_at']

    

    @action(detail=True, methods=['post'])

    def update_stock(self, request, pk=None):

        """更新库存"""

        product = self.get_object()

        quantity = request.data.get('quantity', 0)

        

        try:

            quantity = int(quantity)

            product.stock += quantity

            if product.stock < 0:

                return Response(

                    {'error': '库存不足'},

                    status=status.HTTP_400_BAD_REQUEST

                )

            product.save()

            serializer = self.get_serializer(product)

            return Response(serializer.data)

        except ValueError:

            return Response(

                {'error': '无效的数量'},

                status=status.HTTP_400_BAD_REQUEST

            )



class CategoryViewSet(viewsets.ModelViewSet):

    """分类视图集"""

    queryset = Category.objects.all()

    serializer_class = CategorySerializer

    permission_classes = [IsAuthenticatedOrReadOnly]



class UserViewSet(viewsets.ReadOnlyModelViewSet):

    """用户视图集（只读）"""

    queryset = User.objects.all()

    serializer_class = UserSerializer

    permission_classes = [IsAuthenticated]

    

    @action(detail=False, methods=['get'])

    def me(self, request):

        """获取当前用户信息"""

        serializer = self.get_serializer(request.user)

        return Response(serializer.data)



# urls.py - 路由配置

from django.urls import path, include

from rest_framework.routers import DefaultRouter

from .views import ProductViewSet, CategoryViewSet, UserViewSet



router = DefaultRouter()

router.register(r'products', ProductViewSet, basename='product')

router.register(r'categories', CategoryViewSet, basename='category')

router.register(r'users', UserViewSet, basename='user')



urlpatterns = [

    path('api/v1/', include(router.urls)),

]



# settings.py - 配置CORS

INSTALLED_APPS = [

    # ...

    'corsheaders',

    'rest_framework',

    'django_filters',

]



MIDDLEWARE = [

    # ...

    'corsheaders.middleware.CorsMiddleware',

    'django.middleware.common.CommonMiddleware',

]



# CORS配置

CORS_ALLOWED_ORIGINS = [

    "http://localhost:3000",  # React开发服务器

    "http://localhost:8080",  # Vue开发服务器

]



CORS_ALLOW_CREDENTIALS = True



# REST Framework配置

REST_FRAMEWORK = {

    'DEFAULT_AUTHENTICATION_CLASSES': [

        'rest_framework.authentication.SessionAuthentication',

        'rest_framework.authentication.TokenAuthentication',

    ],

    'DEFAULT_PERMISSION_CLASSES': [

        'rest_framework.permissions.IsAuthenticatedOrReadOnly',

    ],

    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',

    'PAGE_SIZE': 20,

    'DEFAULT_FILTER_BACKENDS': [

        'django_filters.rest_framework.DjangoFilterBackend',

        'rest_framework.filters.SearchFilter',

        'rest_framework.filters.OrderingFilter',

    ],

}

---

39.2 API网关设计

欢迎来到我们企业服务架构中心的第二站——API网关服务中心！这座现代化的服务中心专门负责统一管理所有API请求，就像企业总部的接待大厅，所有访客都需要在这里登记、验证身份，然后被引导到相应的部门。

🚪 API网关的核心功能

路由和负载均衡

API网关的核心功能之一是智能路由和负载均衡：

# 示例4：API网关路由和负载均衡系统

"""

API网关路由和负载均衡实现



包含：

- 路由规则配置

- 负载均衡算法

- 服务发现机制

"""



import random

import time

from typing import List, Dict, Optional

from dataclasses import dataclass

from enum import Enum

from collections import defaultdict



class LoadBalanceAlgorithm(Enum):

    """负载均衡算法枚举"""

    ROUND_ROBIN = "round_robin"  # 轮询

    RANDOM = "random"  # 随机

    WEIGHTED_ROUND_ROBIN = "weighted_round_robin"  # 加权轮询

    LEAST_CONNECTIONS = "least_connections"  # 最少连接

    IP_HASH = "ip_hash"  # IP哈希



@dataclass

class BackendService:

    """后端服务定义"""

    name: str

    url: str

    weight: int = 1

    health: bool = True

    connections: int = 0

    response_time: float = 0.0



class APIGatewayRouter:

    """API网关路由器"""

    

    def __init__(self):

        """初始化路由器"""

        self.routes: Dict[str, List[BackendService]] = {}

        self.algorithm = LoadBalanceAlgorithm.ROUND_ROBIN

        self.round_robin_index = defaultdict(int)

        print("🚪 API网关路由器启动成功！")

    

    def register_route(self, path_pattern: str, services: List[BackendService]):

        """注册路由规则"""

        self.routes[path_pattern] = services

        print(f"✅ 注册路由: {path_pattern} -> {len(services)}个服务")

    

    def route_request(self, path: str, client_ip: str = None) -> Optional[BackendService]:

        """路由请求到后端服务"""

        # 匹配路由规则

        matched_route = None

        for pattern, services in self.routes.items():

            if self._match_pattern(pattern, path):

                matched_route = (pattern, services)

                break

        

        if not matched_route:

            print(f"❌ 未找到匹配的路由: {path}")

            return None

        

        pattern, services = matched_route

        available_services = [s for s in services if s.health]

        

        if not available_services:

            print(f"❌ 没有可用的后端服务: {path}")

            return None

        

        # 根据算法选择服务

        selected_service = self._select_service(available_services, pattern, client_ip)

        

        print(f"📍 路由请求 {path} -> {selected_service.name} ({selected_service.url})")

        return selected_service

    

    def _match_pattern(self, pattern: str, path: str) -> bool:

        """匹配路由模式"""

        # 简单的通配符匹配

        if '*' in pattern:

            pattern_parts = pattern.split('*')

            return all(part in path for part in pattern_parts if part)

        return pattern == path or path.startswith(pattern)

    

    def _select_service(self, services: List[BackendService], 

                       pattern: str, client_ip: str = None) -> BackendService:

        """根据负载均衡算法选择服务"""

        if self.algorithm == LoadBalanceAlgorithm.ROUND_ROBIN:

            index = self.round_robin_index[pattern] % len(services)

            self.round_robin_index[pattern] += 1

            return services[index]

        

        elif self.algorithm == LoadBalanceAlgorithm.RANDOM:

            return random.choice(services)

        

        elif self.algorithm == LoadBalanceAlgorithm.WEIGHTED_ROUND_ROBIN:

            # 加权轮询

            total_weight = sum(s.weight for s in services)

            index = self.round_robin_index[pattern] % total_weight

            self.round_robin_index[pattern] += 1

            

            current_weight = 0

            for service in services:

                current_weight += service.weight

                if index < current_weight:

                    return service

            return services[0]

        

        elif self.algorithm == LoadBalanceAlgorithm.LEAST_CONNECTIONS:

            # 最少连接

            return min(services, key=lambda s: s.connections)

        

        elif self.algorithm == LoadBalanceAlgorithm.IP_HASH:

            # IP哈希

            if client_ip:

                hash_value = hash(client_ip) % len(services)

                return services[hash_value]

            return services[0]

        

        return services[0]

    

    def set_algorithm(self, algorithm: LoadBalanceAlgorithm):

        """设置负载均衡算法"""

        self.algorithm = algorithm

        print(f"🔄 负载均衡算法切换为: {algorithm.value}")



class APIGateway:

    """API网关主类"""

    

    def __init__(self):

        """初始化API网关"""

        self.router = APIGatewayRouter()

        self.request_count = defaultdict(int)

        self.error_count = defaultdict(int)

        print("🌐 API网关启动成功！")

    

    def configure_routes(self):

        """配置路由规则"""

        # 用户服务

        self.router.register_route(

            "/api/v1/users/*",

            [

                BackendService("user-service-1", "http://user-service-1:8000", weight=3),

                BackendService("user-service-2", "http://user-service-2:8000", weight=2),

                BackendService("user-service-3", "http://user-service-3:8000", weight=1),

            ]

        )

        

        # 产品服务

        self.router.register_route(

            "/api/v1/products/*",

            [

                BackendService("product-service-1", "http://product-service-1:8001", weight=2),

                BackendService("product-service-2", "http://product-service-2:8001", weight=2),

            ]

        )

        

        # 订单服务

        self.router.register_route(

            "/api/v1/orders/*",

            [

                BackendService("order-service-1", "http://order-service-1:8002", weight=1),

            ]

        )

    

    def handle_request(self, path: str, method: str = "GET", 

                      client_ip: str = None, **kwargs) -> Dict:

        """处理API请求"""

        self.request_count[path] += 1

        

        # 路由到后端服务

        backend_service = self.router.route_request(path, client_ip)

        

        if not backend_service:

            self.error_count[path] += 1

            return {

                "status": "error",

                "message": "服务不可用",

                "status_code": 503

            }

        

        # 模拟请求处理

        backend_service.connections += 1

        try:

            # 这里应该是实际的HTTP请求

            result = self._forward_request(backend_service, path, method, **kwargs)

            backend_service.connections -= 1

            return result

        except Exception as e:

            backend_service.connections -= 1

            self.error_count[path] += 1

            return {

                "status": "error",

                "message": str(e),

                "status_code": 500

            }

    

    def _forward_request(self, service: BackendService, path: str, 

                        method: str, **kwargs) -> Dict:

        """转发请求到后端服务"""

        # 模拟请求处理

        time.sleep(0.01)  # 模拟网络延迟

        return {

            "status": "success",

            "data": f"Response from {service.name}",

            "status_code": 200

        }

    

    def get_statistics(self) -> Dict:

        """获取网关统计信息"""

        return {

            "total_requests": sum(self.request_count.values()),

            "total_errors": sum(self.error_count.values()),

            "request_by_path": dict(self.request_count),

            "error_by_path": dict(self.error_count)

        }



# 运行演示

if __name__ == "__main__":

    gateway = APIGateway()

    gateway.configure_routes()

    

    # 测试不同负载均衡算法

    algorithms = [

        LoadBalanceAlgorithm.ROUND_ROBIN,

        LoadBalanceAlgorithm.RANDOM,

        LoadBalanceAlgorithm.WEIGHTED_ROUND_ROBIN,

    ]

    

    for algorithm in algorithms:

        print(f"\n{'='*60}")

        print(f"测试负载均衡算法: {algorithm.value}")

        print('='*60)

        gateway.router.set_algorithm(algorithm)

        

        # 模拟10个请求

        for i in range(10):

            gateway.handle_request(

                "/api/v1/users/list",

                client_ip=f"192.168.1.{i % 5}"

            )

    

    # 显示统计信息

    print(f"\n{'='*60}")

    print("网关统计信息")

    print('='*60)

    stats = gateway.get_statistics()

    print(f"总请求数: {stats['total_requests']}")

    print(f"总错误数: {stats['total_errors']}")

限流和熔断机制

API网关还需要实现限流和熔断机制来保护后端服务：

# 示例5：API网关限流和熔断机制

"""

API网关限流和熔断机制实现



包含：

- 令牌桶限流算法

- 熔断器模式

- 降级处理策略

"""



import time

from typing import Dict, Optional

from dataclasses import dataclass, field

from enum import Enum

from collections import deque

from threading import Lock



class CircuitState(Enum):

    """熔断器状态"""

    CLOSED = "closed"  # 关闭（正常）

    OPEN = "open"  # 打开（熔断）

    HALF_OPEN = "half_open"  # 半开（测试）



@dataclass

class RateLimitConfig:

    """限流配置"""

    max_requests: int = 100  # 最大请求数

    time_window: int = 60  # 时间窗口（秒）

    burst_size: int = 10  # 突发容量



@dataclass

class CircuitBreakerConfig:

    """熔断器配置"""

    failure_threshold: int = 5  # 失败阈值

    success_threshold: int = 2  # 成功阈值（半开状态）

    timeout: int = 60  # 超时时间（秒）

    half_open_max_calls: int = 3  # 半开状态最大测试请求数



class TokenBucket:

    """令牌桶限流算法"""

    

    def __init__(self, capacity: int, refill_rate: float):

        """

        初始化令牌桶

        capacity: 桶容量

        refill_rate: 每秒补充的令牌数

        """

        self.capacity = capacity

        self.tokens = capacity

        self.refill_rate = refill_rate

        self.last_refill = time.time()

        self.lock = Lock()

    

    def acquire(self, tokens: int = 1) -> bool:

        """获取令牌"""

        with self.lock:

            self._refill()

            if self.tokens >= tokens:

                self.tokens -= tokens

                return True

            return False

    

    def _refill(self):

        """补充令牌"""

        now = time.time()

        elapsed = now - self.last_refill

        tokens_to_add = elapsed * self.refill_rate

        self.tokens = min(self.capacity, self.tokens + tokens_to_add)

        self.last_refill = now

    

    def get_available_tokens(self) -> int:

        """获取可用令牌数"""

        with self.lock:

            self._refill()

            return int(self.tokens)



class SlidingWindowRateLimiter:

    """滑动窗口限流器"""

    

    def __init__(self, max_requests: int, time_window: int):

        """

        初始化滑动窗口限流器

        max_requests: 最大请求数

        time_window: 时间窗口（秒）

        """

        self.max_requests = max_requests

        self.time_window = time_window

        self.requests = deque()

        self.lock = Lock()

    

    def is_allowed(self, identifier: str = "default") -> bool:

        """检查是否允许请求"""

        with self.lock:

            now = time.time()

            # 移除过期请求

            while self.requests and self.requests[0] < now - self.time_window:

                self.requests.popleft()

            

            # 检查是否超过限制

            if len(self.requests) < self.max_requests:

                self.requests.append(now)

                return True

            return False

    

    def get_remaining(self) -> int:

        """获取剩余请求数"""

        with self.lock:

            now = time.time()

            # 移除过期请求

            while self.requests and self.requests[0] < now - self.time_window:

                self.requests.popleft()

            return max(0, self.max_requests - len(self.requests))



class CircuitBreaker:

    """熔断器"""

    

    def __init__(self, config: CircuitBreakerConfig):

        """初始化熔断器"""

        self.config = config

        self.state = CircuitState.CLOSED

        self.failure_count = 0

        self.success_count = 0

        self.last_failure_time = None

        self.half_open_calls = 0

        self.lock = Lock()

    

    def call(self, func, *args, **kwargs):

        """通过熔断器调用函数"""

        with self.lock:

            if self.state == CircuitState.OPEN:

                # 检查是否可以进入半开状态

                if self._should_attempt_reset():

                    self.state = CircuitState.HALF_OPEN

                    self.half_open_calls = 0

                    print("🔄 熔断器进入半开状态，开始测试")

                else:

                    raise Exception("熔断器打开，请求被拒绝")

            

            elif self.state == CircuitState.HALF_OPEN:

                if self.half_open_calls >= self.config.half_open_max_calls:

                    raise Exception("半开状态测试请求数已达上限")

                self.half_open_calls += 1

        

        # 执行函数

        try:

            result = func(*args, **kwargs)

            self._on_success()

            return result

        except Exception as e:

            self._on_failure()

            raise e

    

    def _on_success(self):

        """处理成功"""

        with self.lock:

            if self.state == CircuitState.HALF_OPEN:

                self.success_count += 1

                if self.success_count >= self.config.success_threshold:

                    self.state = CircuitState.CLOSED

                    self.failure_count = 0

                    self.success_count = 0

                    print("✅ 熔断器关闭，服务恢复正常")

            else:

                self.failure_count = 0

    

    def _on_failure(self):

        """处理失败"""

        with self.lock:

            self.failure_count += 1

            self.last_failure_time = time.time()

            

            if self.state == CircuitState.HALF_OPEN:

                self.state = CircuitState.OPEN

                self.half_open_calls = 0

                print("❌ 半开状态测试失败，熔断器重新打开")

            elif self.failure_count >= self.config.failure_threshold:

                self.state = CircuitState.OPEN

                print(f"🔴 熔断器打开，失败次数: {self.failure_count}")

    

    def _should_attempt_reset(self) -> bool:

        """检查是否应该尝试重置"""

        if self.last_failure_time is None:

            return True

        return time.time() - self.last_failure_time >= self.config.timeout



class RateLimitAndCircuitBreakerGateway:

    """带限流和熔断的API网关"""

    

    def __init__(self):

        """初始化网关"""

        self.rate_limiters: Dict[str, SlidingWindowRateLimiter] = {}

        self.circuit_breakers: Dict[str, CircuitBreaker] = {}

        print("🛡️ 限流和熔断API网关启动成功！")

    

    def add_rate_limiter(self, path: str, max_requests: int, time_window: int):

        """添加限流器"""

        self.rate_limiters[path] = SlidingWindowRateLimiter(max_requests, time_window)

        print(f"✅ 为路径 {path} 添加限流器: {max_requests}请求/{time_window}秒")

    

    def add_circuit_breaker(self, service: str, config: CircuitBreakerConfig):

        """添加熔断器"""

        self.circuit_breakers[service] = CircuitBreaker(config)

        print(f"✅ 为服务 {service} 添加熔断器")

    

    def handle_request(self, path: str, service: str, client_ip: str = None) -> Dict:

        """处理请求（带限流和熔断）"""

        # 限流检查

        limiter = self.rate_limiters.get(path)

        if limiter:

            if not limiter.is_allowed(client_ip or "default"):

                return {

                    "status": "error",

                    "message": "请求频率过高，请稍后再试",

                    "status_code": 429,

                    "remaining": limiter.get_remaining()

                }

        

        # 熔断器检查

        breaker = self.circuit_breakers.get(service)

        if breaker:

            try:

                result = breaker.call(self._call_backend_service, service, path)

                return result

            except Exception as e:

                return {

                    "status": "error",

                    "message": f"服务暂时不可用: {str(e)}",

                    "status_code": 503

                }

        

        # 正常处理

        return self._call_backend_service(service, path)

    

    def _call_backend_service(self, service: str, path: str) -> Dict:

        """调用后端服务（模拟）"""

        # 模拟服务调用

        time.sleep(0.01)

        # 模拟随机失败

        if time.time() % 10 < 1:  # 10%失败率

            raise Exception("后端服务错误")

        return {

            "status": "success",

            "data": f"Response from {service}",

            "status_code": 200

        }



# 运行演示

if __name__ == "__main__":

    gateway = RateLimitAndCircuitBreakerGateway()

    

    # 配置限流

    gateway.add_rate_limiter("/api/v1/users/*", max_requests=10, time_window=60)

    gateway.add_rate_limiter("/api/v1/products/*", max_requests=20, time_window=60)

    

    # 配置熔断器

    gateway.add_circuit_breaker("user-service", CircuitBreakerConfig(

        failure_threshold=3,

        success_threshold=2,

        timeout=30

    ))

    

    # 模拟请求

    print("\n模拟请求处理：")

    for i in range(15):

        result = gateway.handle_request(

            "/api/v1/users/list",

            "user-service",

            client_ip=f"192.168.1.1"

        )

        print(f"请求 {i+1}: {result.get('status')} - {result.get('message', '')}")

        time.sleep(0.1)

服务发现与注册

在微服务架构中，服务发现是API网关的重要功能：

# 示例6：服务发现与注册系统

"""

服务发现与注册实现



包含：

- 服务注册

- 服务发现

- 健康检查

- 服务注销

"""



import time

import threading

from typing import Dict, List, Optional

from dataclasses import dataclass, field

from datetime import datetime, timedelta

from enum import Enum



class ServiceStatus(Enum):

    """服务状态"""

    HEALTHY = "healthy"

    UNHEALTHY = "unhealthy"

    UNKNOWN = "unknown"



@dataclass

class ServiceInstance:

    """服务实例"""

    service_name: str

    instance_id: str

    host: str

    port: int

    metadata: Dict = field(default_factory=dict)

    status: ServiceStatus = ServiceStatus.UNKNOWN

    last_heartbeat: datetime = field(default_factory=datetime.now)

    registered_at: datetime = field(default_factory=datetime.now)



class ServiceRegistry:

    """服务注册中心"""

    

    def __init__(self, heartbeat_timeout: int = 30):

        """初始化注册中心"""

        self.services: Dict[str, List[ServiceInstance]] = {}

        self.heartbeat_timeout = heartbeat_timeout

        self.lock = threading.Lock()

        self.health_check_thread = None

        self.running = False

        print("📋 服务注册中心启动成功！")

    

    def register(self, instance: ServiceInstance) -> bool:

        """注册服务实例"""

        with self.lock:

            if instance.service_name not in self.services:

                self.services[instance.service_name] = []

            

            # 检查是否已存在

            existing = next(

                (s for s in self.services[instance.service_name] 

                 if s.instance_id == instance.instance_id),

                None

            )

            

            if existing:

                # 更新现有实例

                existing.host = instance.host

                existing.port = instance.port

                existing.metadata = instance.metadata

                existing.last_heartbeat = datetime.now()

                existing.status = ServiceStatus.HEALTHY

                print(f"🔄 更新服务实例: {instance.service_name}/{instance.instance_id}")

            else:

                # 添加新实例

                instance.status = ServiceStatus.HEALTHY

                self.services[instance.service_name].append(instance)

                print(f"✅ 注册服务实例: {instance.service_name}/{instance.instance_id}")

            

            return True

    

    def deregister(self, service_name: str, instance_id: str) -> bool:

        """注销服务实例"""

        with self.lock:

            if service_name in self.services:

                self.services[service_name] = [

                    s for s in self.services[service_name]

                    if s.instance_id != instance_id

                ]

                print(f"❌ 注销服务实例: {service_name}/{instance_id}")

                return True

            return False

    

    def discover(self, service_name: str, healthy_only: bool = True) -> List[ServiceInstance]:

        """发现服务实例"""

        with self.lock:

            instances = self.services.get(service_name, [])

            if healthy_only:

                instances = [s for s in instances if s.status == ServiceStatus.HEALTHY]

            return instances.copy()

    

    def heartbeat(self, service_name: str, instance_id: str) -> bool:

        """接收心跳"""

        with self.lock:

            instances = self.services.get(service_name, [])

            for instance in instances:

                if instance.instance_id == instance_id:

                    instance.last_heartbeat = datetime.now()

                    instance.status = ServiceStatus.HEALTHY

                    return True

            return False

    

    def start_health_check(self):

        """启动健康检查"""

        self.running = True

        self.health_check_thread = threading.Thread(target=self._health_check_loop, daemon=True)

        self.health_check_thread.start()

        print("🏥 健康检查线程启动")

    

    def stop_health_check(self):

        """停止健康检查"""

        self.running = False

        if self.health_check_thread:

            self.health_check_thread.join()

        print("🛑 健康检查线程停止")

    

    def _health_check_loop(self):

        """健康检查循环"""

        while self.running:

            time.sleep(5)  # 每5秒检查一次

            with self.lock:

                now = datetime.now()

                for service_name, instances in self.services.items():

                    for instance in instances:

                        elapsed = (now - instance.last_heartbeat).total_seconds()

                        if elapsed > self.heartbeat_timeout:

                            instance.status = ServiceStatus.UNHEALTHY

                            print(f"⚠️ 服务实例不健康: {service_name}/{instance.instance_id}")



# 运行演示

if __name__ == "__main__":

    registry = ServiceRegistry()

    registry.start_health_check()

    

    # 注册服务

    registry.register(ServiceInstance(

        service_name="user-service",

        instance_id="user-1",

        host="192.168.1.10",

        port=8000,

        metadata={"version": "1.0.0", "region": "us-east"}

    ))

    

    registry.register(ServiceInstance(

        service_name="user-service",

        instance_id="user-2",

        host="192.168.1.11",

        port=8000,

        metadata={"version": "1.0.0", "region": "us-west"}

    ))

    

    # 发现服务

    instances = registry.discover("user-service")

    print(f"\n发现 {len(instances)} 个user-service实例")

    for instance in instances:

        print(f"  - {instance.instance_id}: {instance.host}:{instance.port}")

    

    # 模拟心跳

    time.sleep(2)

    registry.heartbeat("user-service", "user-1")

    

    registry.stop_health_check()

---

39.3 跨域和安全处理

欢迎来到我们企业服务架构中心的第三站——安全防护中心！这座现代化的安全堡垒专门负责处理跨域请求和安全防护，就像企业总部的安全部门，确保所有访问都经过严格的身份验证和授权。

🛡️ CORS配置管理

跨域资源共享（CORS）是前后端分离架构中必须处理的问题：

# 示例7：CORS配置管理系统

"""

CORS配置管理实现



包含：

- CORS策略配置

- 预检请求处理

- 跨域请求验证

"""



from typing import List, Optional, Dict

from dataclasses import dataclass

from enum import Enum



class CORSMethod(Enum):

    """HTTP方法枚举"""

    GET = "GET"

    POST = "POST"

    PUT = "PUT"

    DELETE = "DELETE"

    PATCH = "PATCH"

    OPTIONS = "OPTIONS"

    HEAD = "HEAD"



@dataclass

class CORSPolicy:

    """CORS策略配置"""

    allowed_origins: List[str]

    allowed_methods: List[CORSMethod]

    allowed_headers: List[str]

    exposed_headers: List[str] = None

    allow_credentials: bool = True

    max_age: int = 3600  # 预检请求缓存时间（秒）

    

    def __post_init__(self):

        if self.exposed_headers is None:

            self.exposed_headers = []



class CORSHandler:

    """CORS处理器"""

    

    def __init__(self, default_policy: CORSPolicy = None):

        """初始化CORS处理器"""

        self.default_policy = default_policy

        self.path_policies: Dict[str, CORSPolicy] = {}

        print("🛡️ CORS处理器启动成功！")

    

    def set_policy(self, path: str, policy: CORSPolicy):

        """为特定路径设置CORS策略"""

        self.path_policies[path] = policy

        print(f"✅ 为路径 {path} 设置CORS策略")

    

    def handle_preflight(self, origin: str, method: str, 

                        headers: List[str], path: str = "/") -> Dict:

        """处理预检请求（OPTIONS）"""

        policy = self.path_policies.get(path, self.default_policy)

        

        if not policy:

            return {

                "status": "error",

                "message": "CORS策略未配置",

                "status_code": 403

            }

        

        # 检查源

        if origin not in policy.allowed_origins and "*" not in policy.allowed_origins:

            return {

                "status": "error",

                "message": f"源 {origin} 不在允许列表中",

                "status_code": 403

            }

        

        # 检查方法

        try:

            request_method = CORSMethod(method)

            if request_method not in policy.allowed_methods:

                return {

                    "status": "error",

                    "message": f"方法 {method} 不在允许列表中",

                    "status_code": 403

                }

        except ValueError:

            return {

                "status": "error",

                "message": f"无效的HTTP方法: {method}",

                "status_code": 403

            }

        

        # 构建CORS响应头

        cors_headers = {

            "Access-Control-Allow-Origin": origin if origin in policy.allowed_origins else policy.allowed_origins[0],

            "Access-Control-Allow-Methods": ", ".join([m.value for m in policy.allowed_methods]),

            "Access-Control-Allow-Headers": ", ".join(policy.allowed_headers),

            "Access-Control-Max-Age": str(policy.max_age)

        }

        

        if policy.allow_credentials:

            cors_headers["Access-Control-Allow-Credentials"] = "true"

        

        if policy.exposed_headers:

            cors_headers["Access-Control-Expose-Headers"] = ", ".join(policy.exposed_headers)

        

        return {

            "status": "success",

            "status_code": 200,

            "headers": cors_headers

        }

    

    def add_cors_headers(self, origin: str, path: str = "/") -> Dict[str, str]:

        """为响应添加CORS头"""

        policy = self.path_policies.get(path, self.default_policy)

        

        if not policy:

            return {}

        

        # 检查源

        if origin not in policy.allowed_origins and "*" not in policy.allowed_origins:

            return {}

        

        headers = {

            "Access-Control-Allow-Origin": origin if origin in policy.allowed_origins else policy.allowed_origins[0]

        }

        

        if policy.allow_credentials:

            headers["Access-Control-Allow-Credentials"] = "true"

        

        if policy.exposed_headers:

            headers["Access-Control-Expose-Headers"] = ", ".join(policy.exposed_headers)

        

        return headers



# Django CORS配置示例

"""

# settings.py - Django CORS配置

INSTALLED_APPS = [

    # ...

    'corsheaders',

]



MIDDLEWARE = [

    # ...

    'corsheaders.middleware.CorsMiddleware',

    'django.middleware.common.CommonMiddleware',

]



# CORS配置

CORS_ALLOWED_ORIGINS = [

    "http://localhost:3000",

    "http://localhost:8080",

    "https://example.com",

]



CORS_ALLOW_CREDENTIALS = True



CORS_ALLOW_METHODS = [

    'DELETE',

    'GET',

    'OPTIONS',

    'PATCH',

    'POST',

    'PUT',

]



CORS_ALLOW_HEADERS = [

    'accept',

    'accept-encoding',

    'authorization',

    'content-type',

    'dnt',

    'origin',

    'user-agent',

    'x-csrftoken',

    'x-requested-with',

]



CORS_PREFLIGHT_MAX_AGE = 86400

"""



# 运行演示

if __name__ == "__main__":

    # 创建默认CORS策略

    default_policy = CORSPolicy(

        allowed_origins=["http://localhost:3000", "https://example.com"],

        allowed_methods=[CORSMethod.GET, CORSMethod.POST, CORSMethod.PUT, CORSMethod.DELETE],

        allowed_headers=["Content-Type", "Authorization", "X-Requested-With"],

        exposed_headers=["X-Total-Count"],

        allow_credentials=True,

        max_age=3600

    )

    

    cors_handler = CORSHandler(default_policy)

    

    # 处理预检请求

    result = cors_handler.handle_preflight(

        origin="http://localhost:3000",

        method="POST",

        headers=["Content-Type", "Authorization"],

        path="/api/v1/users"

    )

    

    print("预检请求处理结果：")

    print(f"状态: {result['status']}")

    if 'headers' in result:

        print("CORS响应头：")

        for key, value in result['headers'].items():

            print(f"  {key}: {value}")

JWT令牌机制

JWT（JSON Web Token）是实现无状态认证的标准方案：

# 示例8：JWT令牌认证系统

"""

JWT令牌认证实现



包含：

- Token生成

- Token验证

- 刷新令牌机制

- 权限控制

"""



import jwt

import time

from datetime import datetime, timedelta

from typing import Dict, Optional, List

from dataclasses import dataclass

from enum import Enum



class TokenType(Enum):

    """令牌类型"""

    ACCESS = "access"

    REFRESH = "refresh"



@dataclass

class TokenPayload:

    """令牌载荷"""

    user_id: int

    username: str

    roles: List[str]

    token_type: TokenType

    exp: datetime

    iat: datetime



class JWTAuthService:

    """JWT认证服务"""

    

    def __init__(self, secret_key: str, algorithm: str = "HS256"):

        """初始化JWT服务"""

        self.secret_key = secret_key

        self.algorithm = algorithm

        self.access_token_expiry = timedelta(minutes=15)  # 访问令牌15分钟

        self.refresh_token_expiry = timedelta(days=7)  # 刷新令牌7天

        print("🔐 JWT认证服务启动成功！")

    

    def generate_access_token(self, user_id: int, username: str, 

                             roles: List[str] = None) -> str:

        """生成访问令牌"""

        now = datetime.utcnow()

        payload = {

            "user_id": user_id,

            "username": username,

            "roles": roles or [],

            "token_type": TokenType.ACCESS.value,

            "iat": now,

            "exp": now + self.access_token_expiry

        }

        

        token = jwt.encode(payload, self.secret_key, algorithm=self.algorithm)

        print(f"✅ 生成访问令牌: user_id={user_id}, username={username}")

        return token

    

    def generate_refresh_token(self, user_id: int, username: str) -> str:

        """生成刷新令牌"""

        now = datetime.utcnow()

        payload = {

            "user_id": user_id,

            "username": username,

            "token_type": TokenType.REFRESH.value,

            "iat": now,

            "exp": now + self.refresh_token_expiry

        }

        

        token = jwt.encode(payload, self.secret_key, algorithm=self.algorithm)

        print(f"✅ 生成刷新令牌: user_id={user_id}")

        return token

    

    def verify_token(self, token: str, token_type: TokenType = TokenType.ACCESS) -> Optional[Dict]:

        """验证令牌"""

        try:

            payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])

            

            # 检查令牌类型

            if payload.get("token_type") != token_type.value:

                print(f"❌ 令牌类型不匹配: 期望{token_type.value}, 实际{payload.get('token_type')}")

                return None

            

            # 检查是否过期（jwt.decode会自动检查，但我们可以额外验证）

            exp = datetime.utcfromtimestamp(payload["exp"])

            if exp < datetime.utcnow():

                print("❌ 令牌已过期")

                return None

            

            print(f"✅ 令牌验证成功: user_id={payload.get('user_id')}")

            return payload

            

        except jwt.ExpiredSignatureError:

            print("❌ 令牌已过期")

            return None

        except jwt.InvalidTokenError as e:

            print(f"❌ 无效令牌: {str(e)}")

            return None

    

    def refresh_access_token(self, refresh_token: str) -> Optional[str]:

        """使用刷新令牌获取新的访问令牌"""

        payload = self.verify_token(refresh_token, TokenType.REFRESH)

        

        if not payload:

            return None

        

        # 生成新的访问令牌

        return self.generate_access_token(

            payload["user_id"],

            payload["username"],

            payload.get("roles", [])

        )

    

    def extract_user_info(self, token: str) -> Optional[Dict]:

        """从令牌中提取用户信息"""

        payload = self.verify_token(token)

        if payload:

            return {

                "user_id": payload.get("user_id"),

                "username": payload.get("username"),

                "roles": payload.get("roles", [])

            }

        return None



# Django JWT认证实现示例

"""

# authentication.py - 自定义JWT认证类

from rest_framework.authentication import BaseAuthentication

from rest_framework.exceptions import AuthenticationFailed

from .jwt_service import JWTAuthService



class JWTAuthentication(BaseAuthentication):

    def __init__(self):

        self.jwt_service = JWTAuthService(secret_key=settings.SECRET_KEY)

    

    def authenticate(self, request):

        auth_header = request.META.get('HTTP_AUTHORIZATION', '')

        

        if not auth_header.startswith('Bearer '):

            return None

        

        token = auth_header.split(' ')[1]

        payload = self.jwt_service.verify_token(token)

        

        if not payload:

            raise AuthenticationFailed('无效的令牌')

        

        # 获取用户对象

        from django.contrib.auth import get_user_model

        User = get_user_model()

        try:

            user = User.objects.get(id=payload['user_id'])

            return (user, token)

        except User.DoesNotExist:

            raise AuthenticationFailed('用户不存在')



# views.py - 登录视图

from rest_framework.views import APIView

from rest_framework.response import Response

from rest_framework import status

from django.contrib.auth import authenticate

from .jwt_service import JWTAuthService



class LoginView(APIView):

    def post(self, request):

        username = request.data.get('username')

        password = request.data.get('password')

        

        user = authenticate(username=username, password=password)

        if not user:

            return Response(

                {'error': '用户名或密码错误'},

                status=status.HTTP_401_UNAUTHORIZED

            )

        

        jwt_service = JWTAuthService(secret_key=settings.SECRET_KEY)

        access_token = jwt_service.generate_access_token(

            user.id, user.username, [g.name for g in user.groups.all()]

        )

        refresh_token = jwt_service.generate_refresh_token(user.id, user.username)

        

        return Response({

            'access_token': access_token,

            'refresh_token': refresh_token,

            'token_type': 'Bearer',

            'expires_in': 900  # 15分钟

        })



class RefreshTokenView(APIView):

    def post(self, request):

        refresh_token = request.data.get('refresh_token')

        

        jwt_service = JWTAuthService(secret_key=settings.SECRET_KEY)

        access_token = jwt_service.refresh_access_token(refresh_token)

        

        if not access_token:

            return Response(

                {'error': '无效的刷新令牌'},

                status=status.HTTP_401_UNAUTHORIZED

            )

        

        return Response({

            'access_token': access_token,

            'token_type': 'Bearer',

            'expires_in': 900

        })

"""



# 运行演示

if __name__ == "__main__":

    jwt_service = JWTAuthService(secret_key="your-secret-key-here")

    

    # 生成令牌

    access_token = jwt_service.generate_access_token(

        user_id=1,

        username="alice",

        roles=["admin", "user"]

    )

    

    refresh_token = jwt_service.generate_refresh_token(

        user_id=1,

        username="alice"

    )

    

    print(f"\n访问令牌: {access_token[:50]}...")

    print(f"刷新令牌: {refresh_token[:50]}...")

    

    # 验证令牌

    payload = jwt_service.verify_token(access_token)

    if payload:

        print(f"\n令牌载荷: {payload}")

    

    # 提取用户信息

    user_info = jwt_service.extract_user_info(access_token)

    print(f"\n用户信息: {user_info}")

    

    # 刷新令牌

    new_access_token = jwt_service.refresh_access_token(refresh_token)

    if new_access_token:

        print(f"\n新访问令牌: {new_access_token[:50]}...")

API安全防护

API安全防护包括输入验证、SQL注入防护、XSS防护等：

# 示例9：API安全防护系统

"""

API安全防护实现



包含：

- 输入验证

- SQL注入防护

- XSS防护

- CSRF防护

"""



import re

import html

from typing import Any, Dict, List

from urllib.parse import quote, unquote



class InputValidator:

    """输入验证器"""

    

    @staticmethod

    def validate_email(email: str) -> bool:

        """验证邮箱格式"""

        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'

        return bool(re.match(pattern, email))

    

    @staticmethod

    def validate_phone(phone: str) -> bool:

        """验证手机号格式（中国）"""

        pattern = r'^1[3-9]\d{9}$'

        return bool(re.match(pattern, phone))

    

    @staticmethod

    def validate_url(url: str) -> bool:

        """验证URL格式"""

        pattern = r'^https?://[^\s/$.?#].[^\s]*$'

        return bool(re.match(pattern, url))

    

    @staticmethod

    def sanitize_string(value: str, max_length: int = 1000) -> str:

        """清理字符串输入"""

        # 移除前后空格

        value = value.strip()

        

        # 限制长度

        if len(value) > max_length:

            value = value[:max_length]

        

        # HTML转义（防止XSS）

        value = html.escape(value)

        

        return value

    

    @staticmethod

    def validate_sql_injection(value: str) -> bool:

        """检测SQL注入攻击"""

        # SQL注入常见模式

        sql_patterns = [

            r"(\b(SELECT|INSERT|UPDATE|DELETE|DROP|CREATE|ALTER|EXEC|EXECUTE)\b)",

            r"(--|#|/\*|\*/)",

            r"(\b(OR|AND)\s+\d+\s*=\s*\d+)",

            r"('|(\\')|(;)|(\\)|(\|))",

        ]

        

        for pattern in sql_patterns:

            if re.search(pattern, value, re.IGNORECASE):

                return False

        return True



class XSSProtection:

    """XSS防护"""

    

    @staticmethod

    def sanitize_html(html_content: str) -> str:

        """清理HTML内容"""

        # HTML转义

        sanitized = html.escape(html_content)

        return sanitized

    

    @staticmethod

    def validate_xss(value: str) -> bool:

        """检测XSS攻击"""

        # XSS常见模式

        xss_patterns = [

            r"<script[^>]*>.*?</script>",

            r"javascript:",

            r"on\w+\s*=",  # 事件处理器

            r"<iframe[^>]*>",

            r"<object[^>]*>",

            r"<embed[^>]*>",

        ]

        

        for pattern in xss_patterns:

            if re.search(pattern, value, re.IGNORECASE):

                return False

        return True



class APISecurityMiddleware:

    """API安全中间件"""

    

    def __init__(self):

        """初始化安全中间件"""

        self.validator = InputValidator()

        self.xss_protection = XSSProtection()

        print("🛡️ API安全中间件启动成功！")

    

    def validate_request(self, data: Dict[str, Any]) -> Dict[str, Any]:

        """验证请求数据"""

        errors = []

        sanitized_data = {}

        

        for key, value in data.items():

            if isinstance(value, str):

                # SQL注入检测

                if not self.validator.validate_sql_injection(value):

                    errors.append(f"字段 {key} 包含潜在的SQL注入攻击")

                    continue

                

                # XSS检测

                if not self.xss_protection.validate_xss(value):

                    errors.append(f"字段 {key} 包含潜在的XSS攻击")

                    continue

                

                # 清理输入

                sanitized_data[key] = self.validator.sanitize_string(value)

            else:

                sanitized_data[key] = value

        

        return {

            "valid": len(errors) == 0,

            "errors": errors,

            "data": sanitized_data

        }

    

    def validate_email_field(self, email: str) -> bool:

        """验证邮箱字段"""

        return self.validator.validate_email(email)

    

    def validate_phone_field(self, phone: str) -> bool:

        """验证手机号字段"""

        return self.validator.validate_phone(phone)



# Django安全配置示例

"""

# settings.py - Django安全配置

SECURE_SSL_REDIRECT = True  # 强制HTTPS

SESSION_COOKIE_SECURE = True  # 安全Cookie

CSRF_COOKIE_SECURE = True  # CSRF安全Cookie

SECURE_BROWSER_XSS_FILTER = True  # XSS过滤器

SECURE_CONTENT_TYPE_NOSNIFF = True  # 内容类型嗅探防护

X_FRAME_OPTIONS = 'DENY'  # 防止点击劫持



# 中间件配置

MIDDLEWARE = [

    'django.middleware.security.SecurityMiddleware',

    'django.middleware.csrf.CsrfViewMiddleware',

    # ...

]

"""



# 运行演示

if __name__ == "__main__":

    security = APISecurityMiddleware()

    

    # 测试输入验证

    test_data = {

        "username": "alice",

        "email": "alice@example.com",

        "phone": "13800138000",

        "description": "<script>alert('XSS')</script>",

        "query": "'; DROP TABLE users; --"

    }

    

    result = security.validate_request(test_data)

    print("请求验证结果：")

    print(f"有效: {result['valid']}")

    if result['errors']:

        print("错误：")

        for error in result['errors']:

            print(f"  - {error}")

    print(f"清理后的数据: {result['data']}")

    

    # 测试邮箱验证

    print(f"\n邮箱验证: {security.validate_email_field('alice@example.com')}")

    print(f"手机号验证: {security.validate_phone_field('13800138000')}")

---

39.4 综合项目：微服务API网关

在本章的最后，我们将综合运用所学的所有技术，构建一个完整的微服务API网关系统。这个系统将整合前后端分离架构、API网关设计、跨域处理、安全防护等所有功能。

项目概述

项目名称：企业级微服务API网关平台

项目目标：

• 实现统一的API入口和路由管理
• 提供完善的认证授权机制
• 实现智能的负载均衡和限流
• 提供全面的监控和日志功能
• 确保API的安全性和高可用性

技术栈：

• Django + Django REST Framework
• Redis（缓存和限流）
• PostgreSQL（配置存储）
• JWT（认证）
• Nginx（反向代理）

项目架构设计

# 示例10：微服务API网关完整实现

"""

微服务API网关完整系统



包含：

- 网关核心服务

- 路由管理

- 认证授权

- 限流熔断

- 监控日志

"""



import json

import time

import logging

from typing import Dict, List, Optional, Any

from dataclasses import dataclass, field

from datetime import datetime

from collections import defaultdict

from threading import Lock



# 配置日志

logging.basicConfig(

    level=logging.INFO,

    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'

)

logger = logging.getLogger(__name__)



@dataclass

class GatewayConfig:

    """网关配置"""

    name: str

    version: str

    max_connections: int = 1000

    request_timeout: int = 30

    enable_rate_limit: bool = True

    enable_circuit_breaker: bool = True

    enable_monitoring: bool = True



@dataclass

class RouteConfig:

    """路由配置"""

    path: str

    service_name: str

    service_urls: List[str]

    methods: List[str]

    auth_required: bool = True

    rate_limit: int = 100

    rate_window: int = 60



class MicroserviceAPIGateway:

    """微服务API网关主类"""

    

    def __init__(self, config: GatewayConfig):

        """初始化网关"""

        self.config = config

        self.routes: Dict[str, RouteConfig] = {}

        self.request_count = defaultdict(int)

        self.error_count = defaultdict(int)

        self.response_times = defaultdict(list)

        self.lock = Lock()

        

        # 注意：这里需要导入之前定义的组件

        # 在实际项目中，这些组件应该从相应的模块导入

        # 为了演示，我们假设这些组件已经定义

        # self.router = APIGatewayRouter()

        # self.rate_limiter = RateLimitAndCircuitBreakerGateway()

        # self.service_registry = ServiceRegistry()

        # self.cors_handler = CORSHandler()

        # self.jwt_service = JWTAuthService(secret_key="gateway-secret-key")

        

        logger.info(f"🚀 微服务API网关启动: {config.name} v{config.version}")

    

    def register_route(self, route_config: RouteConfig):

        """注册路由"""

        with self.lock:

            self.routes[route_config.path] = route_config

            

            # 注册到路由器

            # 注意：BackendService 和 router 需要从相应模块导入

            # 这里为了演示，假设已经定义

            # backend_services = [

            #     BackendService(

            #         name=f"{route_config.service_name}-{i}",

            #         url=url,

            #         weight=1

            #     )

            #     for i, url in enumerate(route_config.service_urls)

            # ]

            # self.router.register_route(route_config.path, backend_services)

            

            # 配置限流

            if self.config.enable_rate_limit:

                self.rate_limiter.add_rate_limiter(

                    route_config.path,

                    route_config.rate_limit,

                    route_config.rate_window

                )

            

            logger.info(f"✅ 注册路由: {route_config.path} -> {route_config.service_name}")

    

    def handle_request(self, path: str, method: str, headers: Dict,

                      body: Dict = None, client_ip: str = None) -> Dict:

        """处理API请求"""

        start_time = time.time()

        

        try:

            # 1. 路由匹配

            route_config = self.routes.get(path)

            if not route_config:

                return self._error_response(404, "路由不存在")

            

            # 2. 方法验证

            if method not in route_config.methods:

                return self._error_response(405, f"方法 {method} 不允许")

            

            # 3. 认证验证

            if route_config.auth_required:

                auth_result = self._authenticate_request(headers)

                if not auth_result["valid"]:

                    return self._error_response(401, auth_result["message"])

            

            # 4. 限流检查

            if self.config.enable_rate_limit:

                rate_limit_result = self.rate_limiter.handle_request(

                    path, route_config.service_name, client_ip

                )

                if rate_limit_result.get("status") == "error":

                    return rate_limit_result

            

            # 5. CORS处理

            origin = headers.get("Origin")

            if origin:

                cors_headers = self.cors_handler.add_cors_headers(origin, path)

            else:

                cors_headers = {}

            

            # 6. 路由到后端服务

            # 注意：这里需要实际的router实现

            # backend_service = self.router.route_request(path, client_ip)

            # if not backend_service:

            #     return self._error_response(503, "后端服务不可用")

            

            # 7. 转发请求（模拟）

            # 在实际项目中，这里应该转发到真实的后端服务

            response = {

                "status": "success",

                "status_code": 200,

                "data": {

                    "message": f"Response from {route_config.service_name}",

                    "path": path,

                    "method": method

                },

                "headers": {}

            }

            

            # 8. 记录统计

            elapsed_time = time.time() - start_time

            self._record_statistics(path, method, elapsed_time, response.get("status_code", 200))

            

            # 9. 添加CORS头

            if cors_headers:

                response["headers"] = {**response.get("headers", {}), **cors_headers}

            

            return response

            

        except Exception as e:

            logger.error(f"处理请求失败: {str(e)}")

            elapsed_time = time.time() - start_time

            self._record_statistics(path, method, elapsed_time, 500)

            return self._error_response(500, f"内部服务器错误: {str(e)}")

    

    def _authenticate_request(self, headers: Dict) -> Dict:

        """认证请求"""

        auth_header = headers.get("Authorization", "")

        

        if not auth_header.startswith("Bearer "):

            return {"valid": False, "message": "缺少认证令牌"}

        

        token = auth_header.split(" ")[1]

        payload = self.jwt_service.verify_token(token)

        

        if not payload:

            return {"valid": False, "message": "无效的认证令牌"}

        

        return {"valid": True, "user_info": payload}

    

    

    def _error_response(self, status_code: int, message: str) -> Dict:

        """生成错误响应"""

        return {

            "status": "error",

            "status_code": status_code,

            "message": message,

            "timestamp": datetime.now().isoformat()

        }

    

    def _record_statistics(self, path: str, method: str, 

                          elapsed_time: float, status_code: int):

        """记录统计信息"""

        with self.lock:

            self.request_count[path] += 1

            self.response_times[path].append(elapsed_time)

            

            if status_code >= 400:

                self.error_count[path] += 1

    

    def get_statistics(self) -> Dict:

        """获取统计信息"""

        with self.lock:

            stats = {

                "total_requests": sum(self.request_count.values()),

                "total_errors": sum(self.error_count.values()),

                "routes": {}

            }

            

            for path in self.routes.keys():

                request_count = self.request_count.get(path, 0)

                error_count = self.error_count.get(path, 0)

                response_times = self.response_times.get(path, [])

                

                avg_response_time = (

                    sum(response_times) / len(response_times)

                    if response_times else 0

                )

                

                stats["routes"][path] = {

                    "requests": request_count,

                    "errors": error_count,

                    "error_rate": error_count / request_count if request_count > 0 else 0,

                    "avg_response_time": avg_response_time

                }

            

            return stats



# 运行演示

if __name__ == "__main__":

    # 创建网关配置

    gateway_config = GatewayConfig(

        name="Enterprise API Gateway",

        version="1.0.0",

        max_connections=1000,

        enable_rate_limit=True,

        enable_circuit_breaker=True,

        enable_monitoring=True

    )

    

    # 创建网关实例

    gateway = MicroserviceAPIGateway(gateway_config)

    

    # 注册路由

    gateway.register_route(RouteConfig(

        path="/api/v1/users/*",

        service_name="user-service",

        service_urls=[

            "http://user-service-1:8000",

            "http://user-service-2:8000"

        ],

        methods=["GET", "POST", "PUT", "DELETE"],

        auth_required=True,

        rate_limit=100,

        rate_window=60

    ))

    

    gateway.register_route(RouteConfig(

        path="/api/v1/products/*",

        service_name="product-service",

        service_urls=["http://product-service:8001"],

        methods=["GET", "POST"],

        auth_required=True,

        rate_limit=200,

        rate_window=60

    ))

    

    # 模拟请求

    print("\n模拟API请求：")

    for i in range(5):

        response = gateway.handle_request(

            path="/api/v1/users/list",

            method="GET",

            headers={

                "Authorization": "Bearer valid_token",

                "Origin": "http://localhost:3000"

            },

            client_ip=f"192.168.1.{i}"

        )

        print(f"请求 {i+1}: {response.get('status')} - {response.get('status_code')}")

    

    # 显示统计信息

    print("\n网关统计信息：")

    stats = gateway.get_statistics()

    print(json.dumps(stats, indent=2, ensure_ascii=False))

---

💡 代码示例（可运行）

示例1：前后端分离架构演示

# 运行示例1的代码

demo = FrontendBackendSeparationDemo()

demo.compare_architectures()

demo.demonstrate_api_contract()

运行结果：

🚀 前后端分离架构演示中心启动成功！

============================================================
📊 传统架构 vs 前后端分离架构对比分析
============================================================
...

示例2：API网关路由系统

# 运行示例4的代码

gateway = APIGateway()

gateway.configure_routes()

gateway.handle_request("/api/v1/users/list")

运行结果：

🌐 API网关启动成功！
✅ 注册路由: /api/v1/users/* -> 3个服务
📍 路由请求 /api/v1/users/list -> user-service-1 (http://user-service-1:8000)

---

🎯 实践练习

基础练习

练习1：实现简单的API路由

创建一个简单的API路由器，能够根据路径将请求路由到不同的后端服务。

# 练习代码框架

class SimpleRouter:

    def __init__(self):

        # TODO: 初始化路由表

        pass

    

    def add_route(self, path, service_url):

        # TODO: 添加路由规则

        pass

    

    def route(self, path):

        # TODO: 根据路径返回对应的服务URL

        pass

练习2：实现JWT令牌生成和验证

实现一个简单的JWT令牌生成和验证功能。

# 练习代码框架

class SimpleJWT:

    def __init__(self, secret_key):

        # TODO: 初始化JWT服务

        pass

    

    def generate_token(self, user_id, username):

        # TODO: 生成JWT令牌

        pass

    

    def verify_token(self, token):

        # TODO: 验证JWT令牌

        pass

中级练习

练习3：实现令牌桶限流算法

实现一个完整的令牌桶限流算法，支持配置桶容量和补充速率。

# 练习代码框架

class TokenBucketLimiter:

    def __init__(self, capacity, refill_rate):

        # TODO: 初始化令牌桶

        pass

    

    def acquire(self, tokens=1):

        # TODO: 获取令牌

        pass

    

    def get_available_tokens(self):

        # TODO: 获取可用令牌数

        pass

练习4：实现简单的熔断器

实现一个基本的熔断器，支持关闭、打开、半开三种状态。

# 练习代码框架

class SimpleCircuitBreaker:

    def __init__(self, failure_threshold, timeout):

        # TODO: 初始化熔断器

        pass

    

    def call(self, func, *args, **kwargs):

        # TODO: 通过熔断器调用函数

        pass

挑战练习

练习5：构建完整的API网关

综合运用本章所学知识，构建一个功能完整的API网关系统，包括：

• 路由管理
• 认证授权
• 限流熔断
• 监控统计

# 练习代码框架

class CompleteAPIGateway:

    def __init__(self):

        # TODO: 初始化网关组件

        pass

    

    def register_service(self, service_config):

        # TODO: 注册服务

        pass

    

    def handle_request(self, request):

        # TODO: 处理请求

        pass

    

    def get_metrics(self):

        # TODO: 获取监控指标

        pass

---

🤔 本章思考题

1. 概念理解题

1. 前后端分离架构的优势和挑战是什么？

   • 请分析前后端分离架构相比传统架构的优势
   • 讨论在实际项目中可能遇到的挑战和解决方案

2. API网关在微服务架构中的作用是什么？

   • 解释API网关的核心功能
   • 分析API网关如何简化微服务架构的复杂性

3. JWT令牌机制相比传统Session机制的优势是什么？

   • 对比JWT和Session的优缺点
   • 讨论在不同场景下的选择策略

2. 应用分析题

1. 如何设计一个高可用的API网关？

   • 分析高可用性的关键要素
   • 设计容错和故障恢复机制

2. 在前后端分离架构中，如何处理状态管理？

   • 分析前端状态管理和后端状态管理的策略
   • 讨论Token机制在状态管理中的作用

3. 如何实现API网关的智能路由？

   • 设计基于负载、地理位置、用户类型的路由策略
   • 分析不同路由算法的适用场景

3. 编程实践题

1. 实现一个支持多种负载均衡算法的API网关

   • 实现轮询、随机、加权轮询、最少连接等算法
   • 对比不同算法在不同场景下的性能

2. 设计一个完整的认证授权系统

   • 实现JWT令牌生成、验证、刷新机制
   • 实现基于角色的权限控制（RBAC）

3. 构建一个API网关监控系统

   • 实现请求统计、错误追踪、性能监控
   • 设计监控数据的存储和展示方案

---

📖 拓展阅读

在线资源

1. Django REST Framework官方文档

   • https://www.django-rest-framework.org/
   • 深入学习DRF的高级特性和最佳实践

2. API网关设计模式

   • https://microservices.io/patterns/apigateway.html
   • 了解API网关的设计模式和最佳实践

3. JWT官方规范

   • https://jwt.io/
   • 理解JWT的完整规范和实现细节

4. CORS规范文档

   • https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS
   • 深入理解跨域资源共享机制

推荐书籍

1. 《微服务架构设计模式》

   • 作者：Chris Richardson
   • 深入讲解微服务架构和API网关设计

2. 《RESTful Web APIs》

   • 作者：Leonard Richardson, Mike Amundsen
   • 学习RESTful API设计的最佳实践

3. 《Web安全深度剖析》

   • 作者：张炳帅
   • 全面了解Web安全防护技术

开源项目

1. Kong API Gateway

   • https://github.com/Kong/kong
   • 学习企业级API网关的实现

2. Django CORS Headers

   • https://github.com/adamchainz/django-cors-headers
   • 学习Django中CORS处理的实现

3. djangorestframework-simplejwt

   • https://github.com/jazzband/djangorestframework-simplejwt
   • 学习Django中JWT认证的实现

---

📋 本章检查清单

在进入下一章之前，请确保你已经：

理论掌握 ✅

• 理解前后端分离架构的设计原则和优势
• 掌握API网关的核心功能和设计模式
• 理解CORS跨域处理的机制和配置方法
• 掌握JWT令牌认证的原理和实现
• 了解API安全防护的常见攻击和防护措施
• 理解服务发现和注册的机制

实践能力 ✅

• 能够使用Django REST Framework构建RESTful API
• 能够实现API网关的路由和负载均衡功能
• 能够配置CORS策略处理跨域请求
• 能够实现JWT令牌的生成、验证和刷新
• 能够实现基本的限流和熔断机制
• 能够构建简单的服务注册和发现系统

项目经验 ✅

• 完成前后端分离架构的API设计
• 实现一个功能完整的API网关系统
• 集成JWT认证和权限控制
• 实现API监控和统计功能
• 完成API安全防护的配置和测试

---

下一章预告：第40章《WebSocket与实时通信》将介绍WebSocket协议原理、Django Channels框架使用，以及如何构建实时通信应用，包括聊天系统和实时数据推送功能。