Django微服务架构 - 从单体到分布式系统的演进

📂 所属阶段：第三部分 — 高级主题
🎯 难度等级：专家级
⏰ 预计学习时间：8-10小时
🎒 前置知识：部署最佳实践, 性能优化

微服务架构概述

微服务定义与特点

微服务是将单一应用拆成一组独立进程的小型服务，用轻量级HTTP/REST或gRPC通信，每个服务负责特定业务领域：

按业务组织：避免技术分层，让团队按产品全生命周期自治
独立部署扩展：单个服务挂了不影响全局，热点服务可单独扩容
去中心化：技术栈、数据存储、治理都可灵活选择

适用场景与挑战

适用场景

大型复杂电商、SaaS平台
多团队并行开发（跨城市、跨时区）
需求变化频繁，需要快速独立迭代
历史单体应用技术债务重构

常见挑战

分布式系统调试难、链路追踪缺失
跨服务数据一致性是核心难点
运维成本指数级上升（日志、监控、健康检查等）

服务拆分策略

核心原则

限界上下文（DDD）：明确每个服务的边界，只做自己领域的事
数据所有权：服务完全拥有自己的数据库，不直接跨库读写
单一职责：每个服务只解决一个核心业务问题
高内聚低耦合：服务内部逻辑高度相关，服务间尽量少依赖

电商微服务拆分示例

我们以典型电商平台为例，拆分成以下核心服务：

用户服务：注册/登录、权限、个人信息
商品服务：商品管理、分类、库存
订单服务：订单创建、状态流转
支付服务：支付对接、退款
通知服务：邮件/短信/推送

每个服务都有自己的独立Django应用和数据库：

# 订单服务models.py（只留外部ID关联，不直接跨库ForeignKey）
from django.db import models

class Order(models.Model):
    STATUS_CHOICES = [
        ('pending', '待支付'),
        ('paid', '已支付'),
        ('shipped', '已发货'),
    ]
    # 只存外部ID，不直接关联用户/商品表
    user_id = models.IntegerField()
    total_amount = models.DecimalField(max_digits=10, decimal_places=2)
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending')
    created_at = models.DateTimeField(auto_now_add=True)

class OrderItem(models.Model):
    order = models.ForeignKey(Order, on_delete=models.CASCADE, related_name='items')
    product_id = models.IntegerField()
    quantity = models.PositiveIntegerField()
    unit_price = models.DecimalField(max_digits=10, decimal_places=2)

API网关设计

核心功能

API网关是微服务的统一入口，帮前端屏蔽后端服务细节，提供以下功能：

请求路由：根据路径/方法转发到对应微服务
统一认证：JWT/OAuth2验证，把用户信息透传给后端
限流熔断：防止流量洪峰，故障服务自动降级
协议转换：比如把REST转成内部gRPC调用

简易Django API网关实现

# gateway_app/views.py（基于Django REST Framework）
from rest_framework.views import APIView
from rest_framework.response import Response
import requests
from django.conf import settings

class GatewayView(APIView):
    # 路由映射表（实际生产可存入数据库或Consul）
    ROUTE_MAP = {
        '/api/users/': 'http://user-service:8000',
        '/api/products/': 'http://product-service:8000',
        '/api/orders/': 'http://order-service:8000',
    }

    def dispatch(self, request, *args, **kwargs):
        # 匹配路由
        target_url = None
        for prefix, base in self.ROUTE_MAP.items():
            if request.path.startswith(prefix):
                target_url = f"{base}{request.path}"
                break
        if not target_url:
            return Response({'error': 'Route not found'}, status=404)
        
        # 透传请求
        headers = self._filter_headers(request.META)
        try:
            resp = requests.request(
                method=request.method,
                url=target_url,
                headers=headers,
                params=request.query_params,
                data=request.body,
                timeout=30
            )
            return Response(
                resp.json() if resp.content else None,
                status=resp.status_code
            )
        except requests.exceptions.Timeout:
            return Response({'error': 'Service timeout'}, status=504)
        except Exception as e:
            return Response({'error': 'Service unavailable'}, status=503)

    def _filter_headers(self, meta):
        # 过滤掉Django内部的headers，只透传业务相关的
        headers = {}
        for k, v in meta.items():
            if k.startswith('HTTP_'):
                headers[k[5:].replace('_', '-').title()] = v
        return headers

服务通信机制

同步通信（REST/gRPC）

适用于需要即时响应的场景，比如下单前查询用户是否有优惠券。

简化版REST客户端

# user_service_client.py
import requests
from django.conf import settings

class UserServiceClient:
    BASE_URL = settings.USER_SERVICE_URL

    @classmethod
    def get_user(cls, user_id):
        try:
            resp = requests.get(f"{cls.BASE_URL}/api/users/{user_id}/", timeout=5)
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            # 生产中可以加降级逻辑，比如返回默认信息或本地缓存
            return None

异步通信（消息队列）

适用于解耦、最终一致性、提高吞吐量的场景，比如下单后发送通知、扣减库存。

基于Redis的简化版消息队列

# message_queue.py
import redis
import json
import uuid
from datetime import datetime

r = redis.from_url(settings.REDIS_URL)

class MessageQueue:
    @staticmethod
    def publish(event_type, data):
        """发布事件"""
        event = {
            'id': str(uuid.uuid4()),
            'type': event_type,
            'data': data,
            'timestamp': datetime.utcnow().isoformat()
        }
        r.publish('events', json.dumps(event))
        # 也可以存入持久化队列
        r.lpush(f"queue:{event_type}", json.dumps(event))
    
    @staticmethod
    def subscribe(event_type, handler):
        """订阅事件（生产中推荐用Celery或Kafka消费者）"""
        pubsub = r.pubsub()
        pubsub.subscribe('events')
        for msg in pubsub.listen():
            if msg['type'] == 'message':
                event = json.loads(msg['data'])
                if event['type'] == event_type:
                    handler(event)

订单创建后的事件发布示例

# order_service/views.py
from rest_framework.views import APIView
from rest_framework.response import Response
from .models import Order
from .message_queue import MessageQueue

class OrderCreateView(APIView):
    def post(self, request):
        # 1. 本地创建订单
        order = Order.objects.create(
            user_id=request.user.id,
            total_amount=request.data['total_amount'],
            status='pending'
        )
        # 2. 发布订单创建事件
        MessageQueue.publish('order_created', {
            'order_id': order.id,
            'user_id': order.user_id,
            'items': request.data['items']
        })
        return Response({'order_id': order.id}, status=201)

分布式数据管理

一致性模型

在分布式系统中，CAP定理（一致性、可用性、分区容忍性）无法同时满足，我们通常选择最终一致性（BASE理论），核心是接受短暂不一致，但最终会同步。

Saga模式（核心最终一致性方案）

Saga模式通过一系列本地事务+补偿操作实现跨服务一致性：

本地事务：每个服务执行自己的本地操作
事件驱动：用消息队列触发下一个事务
补偿操作：如果某个事务失败，反向执行之前的成功操作

简化版订单创建Saga

# order_service/saga.py
from .models import Order
from .user_service_client import UserServiceClient
from .product_service_client import ProductServiceClient
from .message_queue import MessageQueue

class OrderSaga:
    @staticmethod
    def execute(order_id):
        order = Order.objects.get(id=order_id)
        try:
            # 步骤1：预扣库存（本地事务在商品服务）
            ProductServiceClient.reserve_inventory(order.items)
            # 步骤2：扣减用户余额（本地事务在用户服务）
            UserServiceClient.deduct_balance(order.user_id, order.total_amount)
            # 步骤3：更新订单状态为已支付（本地事务）
            order.status = 'paid'
            order.save()
        except Exception as e:
            # 执行补偿
            OrderSaga.compensate(order_id)
            raise e
    
    @staticmethod
    def compensate(order_id):
        order = Order.objects.get(id=order_id)
        # 1. 释放库存
        ProductServiceClient.release_inventory(order.items)
        # 2. 退款（如果已扣减）
        if order.status == 'paid':
            UserServiceClient.refund_balance(order.user_id, order.total_amount)
        # 3. 更新订单状态为已取消
        order.status = 'cancelled'
        order.save()

迁移策略

绞杀者模式（最稳妥的迁移方式）

不要一次性重写整个单体应用，而是：

新功能用微服务：新开发的功能直接用独立微服务实现
旧功能逐步替换：从边缘功能（比如通知、评价）开始，慢慢替换核心功能
统一入口：用API网关把流量分到新旧系统

迁移步骤建议

分析单体应用：梳理功能模块、依赖关系、数据边界
部署基础设施：先搭好API网关、服务注册发现、监控日志系统
拆分边缘功能：比如通知服务、评价服务，积累经验
拆分核心功能：比如商品服务、订单服务
下线单体应用：所有功能都拆分完成后，再下线旧系统

本章小结

本章我们学习了Django微服务的核心知识点：

服务拆分：基于DDD的限界上下文，每个服务只做自己领域的事，拥有独立数据
API网关：提供统一入口，屏蔽后端服务细节
服务通信：同步用REST/gRPC，异步用消息队列
数据一致性：优先选最终一致性，用Saga模式实现
迁移策略：用绞杀者模式逐步替换，降低风险

💡 核心要点：微服务不是银弹！只有当单体应用确实遇到了瓶颈

#Django微服务架构 - 从单体到分布式系统的演进

#目录

#微服务架构概述

#微服务定义与特点

#适用场景与挑战

#适用场景

#常见挑战

#服务拆分策略

#核心原则

#电商微服务拆分示例

#API网关设计

#核心功能

#简易Django API网关实现

#服务通信机制

#同步通信（REST/gRPC）

#简化版REST客户端

#异步通信（消息队列）

#基于Redis的简化版消息队列

#订单创建后的事件发布示例

#分布式数据管理

#一致性模型

#Saga模式（核心最终一致性方案）

#简化版订单创建Saga

#迁移策略

#绞杀者模式（最稳妥的迁移方式）

#迁移步骤建议

#本章小结