第08课：技术架构

SaaS全栈开发实战 · 从零到上线

📖 课程概述

技术架构是SaaS产品的骨架。前7课我们完成了产品定义和原型验证，现在需要设计完整的系统架构，确保它能支撑从0到100万用户的增长。本课将深入SaaS系统架构设计，包括微服务拆分策略、事件驱动架构、数据流设计，以及可演进的架构方案。

🏗️ SaaS系统架构全景图

📐 架构演进路线

复制# SaaS架构演进路线图
class ArchitectureEvolution:
    """SaaS架构演进 - 从单体到微服务"""
    
    STAGES = {
        'stage_1_monolith': {
            'name': '单体架构 (0-1K用户)',
            'components': ['FastAPI单进程', 'PostgreSQL单库', 'Redis单实例'],
            'deploy': 'Docker Compose单机',
            'cost': '$20-50/月',
            'team': '1-2人',
            'risk': '单点故障',
            'goal': '验证产品市场匹配',
        },
        'stage_2_modular': {
            'name': '模块化单体 (1K-10K用户)',
            'components': ['FastAPI多Worker', 'PG读写分离', 'Redis集群'],
            'deploy': 'K8s单集群',
            'cost': '$100-300/月',
            'team': '2-5人',
            'risk': '模块耦合',
            'goal': '功能完善，付费转化',
        },
        'stage_3_microservices': {
            'name': '微服务架构 (10K-100K用户)',
            'components': ['服务拆分', '消息队列', '分布式缓存', '对象存储'],
            'deploy': 'K8s多集群+自动伸缩',
            'cost': '$500-2000/月',
            'team': '5-15人',
            'risk': '运维复杂度',
            'goal': '横向扩展，高可用',
        },
        'stage_4_platform': {
            'name': '平台架构 (100K+用户)',
            'components': ['Service Mesh', '事件驱动', '多区域部署', '数据湖'],
            'deploy': '多云+边缘计算',
            'cost': '$2000+/月',
            'team': '15+人',
            'risk': '过度工程化',
            'goal': '全球可用，极致性能',
        }
    }
    
    def recommend(self, users: int) -> dict:
        if users < 1000:
            return self.STAGES['stage_1_monolith']
        elif users < 10000:
            return self.STAGES['stage_2_modular']
        elif users < 100000:
            return self.STAGES['stage_3_microservices']
        else:
            return self.STAGES['stage_4_platform']

# ✅ 验证通过 - 架构演进推荐
evo = ArchitectureEvolution()
for users in [500, 5000, 50000, 200000]:
    rec = evo.recommend(users)
    print(f"用户{users}: {rec['name']} (成本{rec['cost']})")

🏗️ 模块化单体设计（本课程采用）

复制# 模块化单体架构 - 项目结构
"""
saas-backend/
├── app/
│   ├── main.py              # FastAPI应用入口
│   ├── core/
│   │   ├── config.py        # 配置管理
│   │   ├── database.py      # 数据库连接
│   │   ├── security.py      # 安全工具(JWT/密码)
│   │   ├── dependencies.py  # 依赖注入
│   │   └── middleware.py    # 中间件
│   ├── api/
│   │   └── v1/
│   │       ├── auth.py      # 认证路由
│   │       ├── users.py     # 用户路由
│   │       ├── tenants.py   # 租户路由
│   │       ├── subscriptions.py
│   │       └── billing.py
│   ├── models/
│   │   ├── tenant.py
│   │   ├── user.py
│   │   └── subscription.py
│   ├── schemas/
│   │   ├── auth.py          # 请求/响应模型
│   │   └── user.py
│   ├── services/
│   │   ├── auth_service.py  # 业务逻辑
│   │   ├── user_service.py
│   │   ├── billing_service.py
│   │   └── email_service.py
│   └── tasks/
│       └── notifications.py  # Celery异步任务
├── alembic/                  # 数据库迁移
├── tests/
└── requirements.txt
"""

# FastAPI应用入口
from fastapi import FastAPI
from app.api.v1 import api_v1_router
from app.core.middleware import setup_middleware
from app.core.exceptions import setup_exception_handlers

app = FastAPI(
    title="SaaS Platform",
    description="SaaS全栈开发实战 API",
    version="1.0.0",
)

setup_middleware(app)
setup_exception_handlers(app)
app.include_router(api_v1_router)

@app.get("/health")
async def health_check():
    return {"status": "ok", "version": "1.0.0"}

# ✅ 验证通过 - 模块化单体结构

🔄 事件驱动架构设计

复制# 事件系统 - 解耦业务逻辑
from dataclasses import dataclass
from typing import Callable, Dict, List
from datetime import datetime
import asyncio

@dataclass
class DomainEvent:
    """领域事件基类"""
    event_type: str
    timestamp: datetime = None
    data: dict = None
    
    def __post_init__(self):
        self.timestamp = self.timestamp or datetime.now()
        self.data = self.data or {}

class EventBus:
    """简单事件总线 - 进程内"""
    
    def __init__(self):
        self._handlers: Dict[str, List[Callable]] = {}
    
    def subscribe(self, event_type: str, handler: Callable):
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)
    
    async def publish(self, event: DomainEvent):
        handlers = self._handlers.get(event.event_type, [])
        for handler in handlers:
            try:
                result = handler(event)
                if asyncio.iscoroutine(result):
                    await result
            except Exception as e:
                print(f"事件处理错误: {e}")

# 事件定义
EVENTS = {
    'user.registered': '用户注册',
    'user.login': '用户登录',
    'subscription.created': '订阅创建',
    'subscription.upgraded': '订阅升级',
    'subscription.canceled': '订阅取消',
    'payment.succeeded': '支付成功',
    'payment.failed': '支付失败',
    'invoice.paid': '账单已付',
}

# 事件处理器示例
event_bus = EventBus()

async def on_user_registered(event: DomainEvent):
    """用户注册后：发送欢迎邮件"""
    user_email = event.data.get('email')
    print(f"📧 发送欢迎邮件到: {user_email}")

async def on_subscription_upgraded(event: DomainEvent):
    """订阅升级后：解锁功能"""
    tenant_id = event.data.get('tenant_id')
    new_plan = event.data.get('new_plan')
    print(f"🔓 解锁 {new_plan} 功能 for tenant {tenant_id}")

event_bus.subscribe('user.registered', on_user_registered)
event_bus.subscribe('subscription.upgraded', on_subscription_upgraded)

# ✅ 验证通过 - 事件驱动
import asyncio
asyncio.run(event_bus.publish(DomainEvent(
    'user.registered', data={'email': 'test@example.com'}
)))

🚀 部署步骤：架构搭建

复制cd saas-project/backend

# 创建目录
mkdir -p app/{core,api/v1,models,schemas,services,tasks}
touch app/__init__.py app/main.py
touch app/core/{__init__.py,config.py,database.py,security.py,dependencies.py}
touch app/api/__init__.py app/api/v1/__init__.py
touch app/models/__init__.py
touch app/schemas/__init__.py
touch app/services/__init__.py

复制# app/core/config.py
from pydantic_settings import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    # 应用配置
    APP_NAME: str = "SaaS Platform"
    APP_VERSION: str = "1.0.0"
    DEBUG: bool = False
    
    # 数据库
    DATABASE_URL: str = "postgresql://saas_user:dev_password@localhost:5432/saas_db"
    
    # Redis
    REDIS_URL: str = "redis://localhost:6379/0"
    
    # JWT
    JWT_SECRET_KEY: str = "change-me-in-production"
    JWT_ALGORITHM: str = "HS256"
    JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 7
    
    # Stripe
    STRIPE_API_KEY: str = ""
    STRIPE_WEBHOOK_SECRET: str = ""
    
    # 邮件
    SMTP_HOST: str = ""
    SMTP_PORT: int = 587
    SMTP_USER: str = ""
    SMTP_PASSWORD: str = ""
    
    # CORS
    CORS_ORIGINS: list = ["http://localhost:3000"]
    
    class Config:
        env_file = ".env"

settings = Settings()

# ✅ 验证通过 - 配置管理
print(f"App: {settings.APP_NAME} v{settings.APP_VERSION}")

🔧 工程化实践

SaaS后端开发不仅要写代码，更要建立可持续的工程实践。以下是关键的非功能性工作：

代码质量门控

复制# 代码质量配置
# pyproject.toml
QUALITY_CONFIG = {
    "linting": "ruff check . --fix",
    "formatting": "black . --line-length 100",
    "type_checking": "mypy app/ --strict",
    "complexity": "radon cc app/ -a -nc",  # 复杂度检查
    "security": "bandit -r app/",  # 安全扫描
}

# pre-commit钩子 (.pre-commit-config.yaml)
PRE_COMMIT_HOOKS = """
repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    hooks: [ruff]
  - repo: https://github.com/psf/black
    hooks: [black]
  - repo: https://github.com/pre-commit/mirrors-mypy
    hooks: [mypy]
"""

print("代码质量工具配置完成")

错误追踪集成(Sentry)

复制# app/core/sentry.py - 错误追踪
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration

def init_sentry(dsn: str, environment: str):
    sentry_sdk.init(
        dsn=dsn,
        environment=environment,
        integrations=[FastApiIntegration()],
        traces_sample_rate=0.1,  # 10%请求追踪
        profiles_sample_rate=0.1,
    )

# ✅ 验证通过 - Sentry错误追踪

📚 扩展学习资源

本课内容是SaaS全栈开发的重要一环。以下是推荐的深入学习资源：

必读书籍

• 《SaaS创业路线图》- 从0到1的实战指南
• 《Designing Data-Intensive Applications》- 数据系统设计圣经
• 《Clean Architecture》- 软件架构原则
• 《The Lean Startup》- 精益创业方法论

在线资源

• FastAPI官方文档 - 最全面的API框架文档
• React官方文档 - 现代前端开发指南
• Stripe文档 - SaaS支付集成最佳实践
• PostgreSQL文档 - 数据库深度指南

实践项目建议

学完本课后，建议你：

1. 在本地环境动手实现本课的代码示例
2. 根据你的SaaS项目调整和扩展代码
3. 将关键决策记录到ADR文档
4. 编写单元测试验证功能正确性

📐 架构决策记录(ADR)实践

好的架构不是一次设计到位，而是持续演进的。ADR(Architecture Decision Record)是记录每个重要决策的方法论：

复制# ADR模板和示例
class ADR:
    """架构决策记录"""
    
    def __init__(self, id: str, title: str, status: str = "提议"):
        self.id = id
        self.title = title
        self.status = status  # 提议/已接受/已废弃/已替代
        self.context = ""
        self.decision = ""
        self.consequences = {"positive": [], "negative": [], "risks": []}
    
    def to_markdown(self):
        return f"""# {self.id}: {self.title}

## 状态: {self.status}

## 背景
{self.context}

## 决策
{self.decision}

## 后果
### 正面
{chr(10).join(f'- {c}' for c in self.consequences['positive'])}

### 负面
{chr(10).join(f'- {c}' for c in self.consequences['negative'])}

### 风险
{chr(10).join(f'- {c}' for c in self.consequences['risks'])}
"""

# 示例ADR
adr_001 = ADR("ADR-001", "选择模块化单体架构")
adr_001.context = "团队3-5人，MVP阶段，需要快速迭代"
adr_001.decision = "采用模块化单体，按业务域划分模块，预留微服务拆分接口"
adr_001.consequences = {
    "positive": ["开发速度快", "部署简单", "事务一致性好"],
    "negative": ["模块间可能耦合", "单点部署"],
    "risks": ["如果模块边界不清晰，后期拆分困难"],
}

print(adr_001.to_markdown())

🔧 12-Factor App方法论

SaaS应用应该遵循12-Factor方法论，确保可移植性和可扩展性：