高并发系统设计实战：从单体到微服务的亿级流量架构演变

引言

大多数系统在起步阶段，都是一个单体应用 + 一台 MySQL，跑得好好的。然后某一天，用户在增长，QPS 从 100 涨到了 1000，系统开始不堪重负。

系统的架构不是设计出来的，是演变出来的。 每个架构决策背后都有一个性能瓶颈和用户投诉。本文模拟一个真实的电商系统，从 0 到 1 演变成一个支撑亿级流量的架构。

第一阶段：单体架构的黄金年代

初始架构

Nginx -> Tomcat(单体应用) -> MySQL

这个架构很简单：

一台 Nginx 做反向代理
一个 Spring Boot 单体应用
一台 MySQL 数据库

第一个性能瓶颈：数据库连接

当用户量开始增长，第一个倒下的一般是数据库。不是 MySQL 不够快，而是数据库连接数不够用。

引入连接池

# HikariCP 配置
spring:
  datasource:
    hikari:
      maximum-pool-size: 20
      minimum-idle: 5
      connection-timeout: 30000
      idle-timeout: 600000
      max-lifetime: 1800000

连接池大小有个经典公式：

connections = ((core_count * 2) + effective_spindle_count)

但实际应用中，数据库连接数一般不超过 CPU 核心数的 2 倍。连接池太大的话，上下文切换的代价会超过并行带来的收益。

第二阶段：引入缓存层

Redis 的出现时机

当你的数据库 QPS 超过 1000 时，就该上 Redis 了。一个简单的 Read-Write Through 模式：

import redis
import json
from typing import Optional, Any
from functools import wraps
import hashlib

class CacheService:
    """
    多级缓存服务：本地缓存 -> Redis -> 数据库
    """

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.local_cache = {}  # 进程内 LRU 缓存

    def cached(self, ttl: int = 300):
        """缓存装饰器"""
        def decorator(func):
            @wraps(func)
            async def wrapper(self, key: str, *args, **kwargs):
                # L1: 本地缓存
                cache_key = f"{func.__name__}:{key}"
                result = self.local_cache.get(cache_key)
                if result is not None:
                    return result

                # L2: Redis
                redis_key = f"cache:{func.__name__}:{key}"
                result = await self.redis.get(redis_key)
                if result is not None:
                    self.local_cache[cache_key] = json.loads(result)
                    return self.local_cache[cache_key]

                # L3: 数据库
                result = await func(self, key, *args, **kwargs)
                if result is not None:
                    await self.redis.setex(
                        redis_key,
                        ttl,
                        json.dumps(result),
                    )
                    self.local_cache[cache_key] = result

                return result
            return wrapper
        return decorator

缓存穿透/击穿/雪崩解决方案

class CacheShield:
    """缓存防护层"""

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    async def get_with_bloom_filter(self, key: str) -> Optional[str]:
        """
        布隆过滤器防止缓存穿透
        - 穿透：请求不存在的数据，每次都打到 DB
        - 解决：用布隆过滤器过滤掉一定不存在的 key
        """
        if not self.bloom_filter.might_contain(key):
            return None  # 一定不存在，不查了

        return await self.redis.get(key)

    async def get_with_mutex(self, key: str) -> Optional[str]:
        """
        互斥锁防止缓存击穿
        - 击穿：热点 key 过期瞬间，大量请求打到 DB
        - 解决：同一时刻只允许一个请求重建缓存
        """
        value = await self.redis.get(key)
        if value is not None:
            return value

        # 分布式锁：只有一个线程能去查 DB
        lock_key = f"lock:rebuild:{key}"
        if await self.redis.set(lock_key, "1", nx=True, ex=10):
            try:
                value = await self._load_from_db(key)
                if value:
                    await self.redis.setex(key, 60, value)
                return value
            finally:
                await self.redis.delete(lock_key)
        else:
            # 获取锁失败，等 50ms 后重试
            await asyncio.sleep(0.05)
            return await self.get_with_mutex(key)

    async def set_with_random_ttl(self, key: str, value: str, base_ttl: int):
        """
        随机过期时间防止缓存雪崩
        - 雪崩：大量 key 同时过期，瞬间流量全打到 DB
        - 解决：过期时间加随机偏移
        """
        import random
        ttl = base_ttl + random.randint(0, base_ttl // 2)
        await self.redis.setex(key, ttl, value)

第三阶段：数据库优化

读写分离

主库(写) -> 从库1(读)、从库2(读)

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from contextlib import contextmanager
import random

class DatabaseRouter:
    """数据库读写分离路由"""

    def __init__(self):
        self.write_engine = create_engine(
            "mysql+pymysql://user:pass@master:3306/db",
            pool_size=10,
            max_overflow=20,
        )
        self.read_engines = [
            create_engine(f"mysql+pymysql://user:pass@slave{i}:3306/db")
            for i in range(2)
        ]
        self.WriteSession = sessionmaker(bind=self.write_engine)

    @contextmanager
    def read_session(self) -> Session:
        """获取读库连接（随机负载均衡）"""
        engine = random.choice(self.read_engines)
        Session = sessionmaker(bind=engine)
        session = Session()
        try:
            yield session
        finally:
            session.close()

    @contextmanager
    def write_session(self) -> Session:
        """获取写库连接"""
        session = self.WriteSession()
        try:
            yield session
            session.commit()
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()

分库分表策略

当单表数据量超过 2000 万行时，MySQL 的写入性能开始显著下降。

Sharding 算法

class ShardingStrategy:
    """一致性哈希分片策略"""

    def __init__(self, shard_count: int = 16):
        self.shard_count = shard_count
        self.virtual_nodes = 128
        self.ring = self._build_ring()

    def get_shard(self, user_id: str) -> int:
        """
        根据用户 ID 确定分片
        使用 user_id % shard_count 保证同一用户的请求落在同一分片
        """
        hash_val = hashlib.md5(user_id.encode()).hexdigest()
        return int(hash_val[:8], 16) % self.shard_count

    def get_table(self, order_id: str) -> int:
        """
        订单表按月分表
        单月数据量可控，方便归档清理
        """
        # order_id 格式: YYYYMMDD_xxxxx
        year_month = order_id[:6]
        return int(year_month)

# 使用示例
class OrderDAO:
    def __init__(self, sharding: ShardingStrategy):
        self.sharding = sharding
        self.shard_engines = {
            i: create_engine(f"mysql://.../db_shard_{i}")
            for i in range(sharding.shard_count)
        }

    def create_order(self, user_id: str, order: dict):
        shard = self.sharding.get_shard(user_id)
        table = f"orders_{self.sharding.get_table(order['order_id'])}"

        engine = self.shard_engines[shard]
        with engine.connect() as conn:
            conn.execute(
                f"INSERT INTO {table} (user_id, product_id, amount) VALUES (:uid, :pid, :amt)",
                uid=user_id,
                pid=order["product_id"],
                amt=order["amount"],
            )

第四阶段：引入消息队列

异步解耦

用户下单 -> 订单服务 -> Kafka -> 库存服务
                     -> Kafka -> 通知服务
                     -> Kafka -> 推荐服务

from confluent_kafka import Producer, Consumer
import json
from dataclasses import dataclass
from typing import Callable

@dataclass
class KafkaConfig:
    bootstrap_servers: str = "localhost:9092"
    group_id: str = "order-service"
    auto_offset_reset: str = "earliest"

class EventBus:
    """基于 Kafka 的事件总线"""

    def __init__(self, config: KafkaConfig):
        self.config = config
        self._producer = None
        self._consumers: dict[str, Consumer] = {}

    @property
    def producer(self):
        if self._producer is None:
            self._producer = Producer({
                'bootstrap.servers': self.config.bootstrap_servers,
                'acks': 'all',  # 等待所有副本确认
                'retries': 3,
                'enable.idempotence': True,  # 幂等性保证
            })
        return self._producer

    def publish(self, topic: str, event: dict, key: str = None):
        """发布事件（异步）"""
        self.producer.produce(
            topic,
            key=key,
            value=json.dumps(event),
            callback=self._delivery_callback,
        )
        self.producer.poll(0)  # 触发回调

    def subscribe(self, topic: str, handler: Callable):
        """订阅事件"""

        consumer = Consumer({
            'bootstrap.servers': self.config.bootstrap_servers,
            'group.id': self.config.group_id,
            'auto.offset.reset': self.config.auto_offset_reset,
            'enable.auto.commit': False,  # 手动提交，保证 at-least-once
        })
        consumer.subscribe([topic])

        # 启动消费循环
        import threading
        thread = threading.Thread(
            target=self._consume_loop,
            args=(consumer, handler),
            daemon=True,
        )
        thread.start()

    def _consume_loop(self, consumer: Consumer, handler: Callable):
        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                continue
            if msg.error():
                print(f"Consumer error: {msg.error()}")
                continue

            try:
                event = json.loads(msg.value())
                handler(event)
                consumer.commit(msg)
            except Exception as e:
                print(f"Handle event error: {e}")
                # 不提交 offset，下次重试

    def _delivery_callback(self, err, msg):
        if err:
            print(f"Message delivery failed: {err}")
        else:
            print(f"Message delivered to {msg.topic()} [{msg.partition()}]")

第五阶段：服务拆分与微服务化

拆服务的方法论

什么时候该拆？一个信号：当你修改一个功能需要改动多个模块时。

# 拆分后的订单服务
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import httpx

app = FastAPI(title="订单服务")

class OrderRequest(BaseModel):
    user_id: str
    product_id: str
    quantity: int

@app.post("/api/orders")
async def create_order(request: OrderRequest):
    # 1. 查询库存（调用库存服务）
    async with httpx.AsyncClient() as client:
        stock_resp = await client.get(
            f"http://inventory-service/api/stock/{request.product_id}"
        )
        if stock_resp.status_code != 200:
            raise HTTPException(status_code=502, detail="库存服务异常")

        stock = stock_resp.json()
        if stock["available"] < request.quantity:
            raise HTTPException(status_code=400, detail="库存不足")

        # 2. 扣减库存
        await client.post(
            f"http://inventory-service/api/stock/deduct",
            json={"product_id": request.product_id, "quantity": request.quantity},
        )

    # 3. 创建订单
    order_id = await create_order_in_db(request)

    # 4. 发送创建事件（异步）
    event_bus.publish("order.created", {
        "order_id": order_id,
        "user_id": request.user_id,
        "product_id": request.product_id,
        "quantity": request.quantity,
    })

    return {"order_id": order_id, "status": "created"}

第六阶段：分布式事务

分布式事务 Saga 模式

from enum import Enum
from typing import List, Callable, Dict
import asyncio

class SagaStatus(Enum):
    PENDING = "pending"
    COMPLETED = "completed"
    COMPENSATING = "compensating"
    FAILED = "failed"

class SagaStep:
    """Saga 事务步骤"""

    def __init__(
        self,
        name: str,
        execute: Callable,
        compensate: Callable,
    ):
        self.name = name
        self.execute = execute
        self.compensate = compensate

class SagaOrchestrator:
    """
    Saga 编排器 - Choreography 模式

    每个步骤执行后，下一个步骤由事件驱动
    如果某步失败，自动补偿之前的所有步骤
    """

    def __init__(self):
        self.steps: List[SagaStep] = []
        self.status = SagaStatus.PENDING
        self.executed_steps: List[SagaStep] = []

    def add_step(self, step: SagaStep):
        self.steps.append(step)

    async def execute(self, context: Dict) -> Dict:
        """执行 Saga 事务"""
        try:
            for step in self.steps:
                print(f"执行步骤: {step.name}")
                result = await step.execute(context)
                self.executed_steps.append(step)
                context.update(result or {})

            self.status = SagaStatus.COMPLETED
            return {"status": "success", "context": context}

        except Exception as e:
            print(f"步骤 {step.name} 执行失败: {e}")
            self.status = SagaStatus.COMPENSATING
            await self._compensate(context)
            self.status = SagaStatus.FAILED
            return {"status": "failed", "error": str(e)}

    async def _compensate(self, context: Dict):
        """回滚已执行的步骤（逆序）"""
        print("开始补偿事务...")
        for step in reversed(self.executed_steps):
            try:
                print(f"补偿步骤: {step.name}")
                await step.compensate(context)
            except Exception as e:
                print(f"补偿步骤 {step.name} 失败: {e}")
                # 补偿失败需要人工介入
                # 发送告警通知运维

# 使用示例：订单创建 Saga
async def create_order_saga(user_id: str, product_id: str, quantity: int):
    saga = SagaOrchestrator()

    # 步骤1：创建订单
    async def execute_create_order(ctx):
        order_id = await order_service.create(ctx["user_id"], ctx["product_id"])
        ctx["order_id"] = order_id
        return ctx

    async def compensate_create_order(ctx):
        await order_service.cancel(ctx["order_id"])

    saga.add_step(SagaStep(
        name="创建订单",
        execute=execute_create_order,
        compensate=compensate_create_order,
    ))

    # 步骤2：扣减库存
    async def execute_deduct_stock(ctx):
        await inventory_service.deduct(ctx["product_id"], ctx["quantity"])
        return ctx

    async def compensate_deduct_stock(ctx):
        await inventory_service.restore(ctx["product_id"], ctx["quantity"])

    saga.add_step(SagaStep(
        name="扣减库存",
        execute=execute_deduct_stock,
        compensate=compensate_deduct_stock,
    ))

    # 步骤3：扣款
    async def execute_charge(ctx):
        await payment_service.charge(ctx["user_id"], ctx["amount"])
        return ctx

    async def compensate_charge(ctx):
        await payment_service.refund(ctx["user_id"], ctx["amount"])

    saga.add_step(SagaStep(
        name="用户扣款",
        execute=execute_charge,
        compensate=compensate_charge,
    ))

    return await saga.execute({
        "user_id": user_id,
        "product_id": product_id,
        "quantity": quantity,
    })

第七阶段：限流与熔断

令牌桶限流

import time
import asyncio
from dataclasses import dataclass

@dataclass
class TokenBucket:
    """
    令牌桶限流器
    允许一定的突发流量（burst），但限制了平均速率
    """
    rate: float           # 每秒生成的令牌数
    capacity: int         # 桶容量（允许的突发流量）
    tokens: float = 0.0
    last_refill: float = 0.0

    def __post_init__(self):
        self.tokens = self.capacity
        self.last_refill = time.monotonic()

    async def acquire(self, tokens: int = 1) -> bool:
        """获取令牌"""
        self._refill()

        if self.tokens >= tokens:
            self.tokens -= tokens
            return True

        return False

    def _refill(self):
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_refill = now


class SentinelGateway:
    """API 网关限流中间件"""

    def __init__(self):
        self.buckets: dict[str, TokenBucket] = {}

    def add_route(self, path: str, rate: int, burst: int):
        """注册路由限流规则"""
        self.buckets[path] = TokenBucket(rate=rate, capacity=burst)

    async def handle(self, path: str, handler):
        """限流中间件"""
        bucket = self.buckets.get(path)
        if bucket is None:
            return await handler()

        if not await bucket.acquire():
            raise HTTPException(
                status_code=429,
                detail="Too Many Requests - 请稍后重试",
                headers={"Retry-After": "5"},
            )

        return await handler()

# 使用示例
gateway = SentinelGateway()
gateway.add_route("/api/orders", rate=100, burst=200)   # 每秒100个请求，允许200突发
gateway.add_route("/api/search", rate=500, burst=800)    # 搜索接口更高频

熔断器模式

from enum import Enum
from datetime import datetime, timedelta
import asyncio

class CircuitState(Enum):
    CLOSED = "closed"            # 正常状态
    OPEN = "open"               # 熔断状态，拒绝请求
    HALF_OPEN = "half_open"     # 半开状态，探测下游是否恢复

class CircuitBreaker:
    """
    熔断器实现

    状态流转：
    CLOSED -> (失败次数达标) -> OPEN -> (冷却时间过了) -> HALF_OPEN
    HALF_OPEN -> (探测成功) -> CLOSED
    HALF_OPEN -> (探测失败) -> OPEN
    """

    def __init__(
        self,
        failure_threshold: int = 5,       # 失败次数阈值
        timeout_seconds: int = 30,        # 冷却时间（秒）
        half_open_max_requests: int = 3,   # 半开状态允许的探测请求数
    ):
        self.failure_threshold = failure_threshold
        self.timeout = timedelta(seconds=timeout_seconds)
        self.half_open_max_requests = half_open_max_requests

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time: datetime = None
        self.half_open_requests = 0

    async def call(self, func, *args, **kwargs):
        """通过熔断器调用下游服务"""
        if self.state == CircuitState.OPEN:
            if self._should_try_half_open():
                self.state = CircuitState.HALF_OPEN
                self.half_open_requests = 0
            else:
                raise CircuitBreakerOpenError("熔断器已打开，请稍后重试")

        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_requests >= self.half_open_max_requests:
                raise CircuitBreakerOpenError("半开状态探测请求数已达上限")

            self.half_open_requests += 1

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

    def _on_success(self):
        """调用成功"""
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED
            self.failure_count = 0

    def _on_failure(self):
        """调用失败"""
        self.failure_count += 1
        self.last_failure_time = datetime.now()

        if (self.state == CircuitState.CLOSED
            and self.failure_count >= self.failure_threshold):
            self.state = CircuitState.OPEN

    def _should_try_half_open(self) -> bool:
        if self.last_failure_time is None:
            return True
        return datetime.now() - self.last_failure_time >= self.timeout

class CircuitBreakerOpenError(Exception):
    pass


# 使用示例
breaker = CircuitBreaker(failure_threshold=3, timeout_seconds=30)

@app.post("/api/orders")
async def create_order(request: OrderRequest):
    try:
        return await breaker.call(
            inventory_service.check_and_deduct,
            request.product_id,
            request.quantity,
        )
    except CircuitBreakerOpenError:
        # 熔断器打开，返回降级响应
        return {
            "status": "degraded",
            "message": "系统繁忙，订单已接收，请稍后查看状态",
            "order_id": await async_create_pending_order(request),
        }

第八阶段：异步非阻塞架构

Python asyncio + FastAPI 的极致优化

import asyncio
from concurrent.futures import ThreadPoolExecutor

class AsyncOptimizer:
    """
    异步优化器 - 避免同步阻塞

    关键原则：
    1. IO 密集用 async/await
    2. CPU 密集用 run_in_executor
    3. 数据库查询用异步驱动（asyncpg / aiomysql）
    """

    def __init__(self):
        self.cpu_executor = ThreadPoolExecutor(max_workers=4)

    async def fetch_user_orders(self, user_id: str) -> dict:
        """并行获取用户数据（多个下游服务同时调）"""
        # 同时调用三个无依赖的服务
        results = await asyncio.gather(
            order_service.get_orders(user_id),
            user_service.get_profile(user_id),
            recommendation_service.get_recommended(user_id),
            return_exceptions=True,  # 部分失败不影响其他
        )

        orders, profile, recommendations = results

        return {
            "user": profile if not isinstance(profile, Exception) else None,
            "orders": orders if not isinstance(orders, Exception) else [],
            "recommendations": recommendations if not isinstance(recommendations, Exception) else [],
        }

    async def compute_heavy_task(self, data: list) -> dict:
        """CPU 密集型任务转移到线程池"""
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(
            self.cpu_executor,
            self._cpu_bound_algorithm,
            data,
        )

    def _cpu_bound_algorithm(self, data: list) -> dict:
        """CPU 密集型计算"""
        # 复杂的数据聚合、计算
        result = {}
        for item in data:
            # ... 复杂运算
            pass
        return result

架构演变总结

阶段	核心问题	解决方案	QPS 上限
单体	数据库连接不足	连接池(HikariCP)	~500
缓存	数据库压力太大	Redis多级缓存	~5,000
数据库	单表数据太大	读写分离 + 分库分表	~20,000
消息队列	同步链路太长	Kafka异步解耦	~50,000
微服务	代码耦合严重	服务拆分 + RPC	~200,000
分布式事务	跨服务数据一致	Saga补偿模式	-
限流熔断	雪崩效应	Sentinel + Hystrix	-
异步架构	CPU/IO阻塞	asyncio + 协程	~500,000

终极原则

不要过早优化：QPS 不到 1000 时，你的瓶颈大概率在你自己的代码，不在架构
先监控后优化：没有监控数据的优化就是盲人摸象
缓存是第一优先级：80% 的性能问题加个 Redis 就解决了
数据一致性 > 性能：分布式系统下，不能拿业务正确性去换 10ms 的延迟提升
架构是长出来的：不要在一开始就设计一个"完美的架构"，跟着业务的节奏一步一步来

声明：本文中的性能数据是基于常见硬件配置的估计值，实际表现会因具体业务逻辑、数据规模和硬件环境而异。架构设计需要结合实际情况进行决策。

高并发系统设计实战：从单体到微服务的亿级流量架构演变

高并发系统设计实战：从单体到微服务的亿级流量架构演变

引言

第一阶段：单体架构的黄金年代

初始架构

第一个性能瓶颈：数据库连接

引入连接池

第二阶段：引入缓存层

Redis 的出现时机

缓存穿透/击穿/雪崩解决方案

第三阶段：数据库优化

读写分离

分库分表策略

Sharding 算法

第四阶段：引入消息队列

异步解耦

第五阶段：服务拆分与微服务化

拆服务的方法论

第六阶段：分布式事务

分布式事务 Saga 模式

第七阶段：限流与熔断

令牌桶限流

熔断器模式

第八阶段：异步非阻塞架构

Python asyncio + FastAPI 的极致优化

架构演变总结

终极原则

💬 评论区

发表评论

高并发系统设计实战：从单体到微服务的亿级流量架构演变#

引言#

第一阶段：单体架构的黄金年代#

初始架构#

第一个性能瓶颈：数据库连接#

引入连接池#

第二阶段：引入缓存层#

Redis 的出现时机#

缓存穿透/击穿/雪崩 解决方案#

第三阶段：数据库优化#

读写分离#

分库分表策略#

Sharding 算法#

第四阶段：引入消息队列#

异步解耦#

第五阶段：服务拆分与微服务化#

拆服务的方法论#

第六阶段：分布式事务#

分布式事务 Saga 模式#

第七阶段：限流与熔断#

令牌桶限流#

熔断器模式#

第八阶段：异步非阻塞架构#

Python asyncio + FastAPI 的极致优化#

架构演变总结#

终极原则#

💬 评论区

发表评论

高并发系统设计实战：从单体到微服务的亿级流量架构演变

引言

第一阶段：单体架构的黄金年代

初始架构

第一个性能瓶颈：数据库连接

引入连接池

第二阶段：引入缓存层

Redis 的出现时机

缓存穿透/击穿/雪崩解决方案

第三阶段：数据库优化

读写分离

分库分表策略

Sharding 算法

第四阶段：引入消息队列

异步解耦

第五阶段：服务拆分与微服务化

拆服务的方法论

第六阶段：分布式事务

分布式事务 Saga 模式

第七阶段：限流与熔断

令牌桶限流

熔断器模式

第八阶段：异步非阻塞架构

Python asyncio + FastAPI 的极致优化

架构演变总结

终极原则