基于Python异步Web框架的全栈性能优化实战：从FastAPI到Nginx+Redis部署

shortlink/
├── app/
│   ├── api/
│   │   ├── v1/
│   │   │   ├── endpoints/
│   │   │   │   ├── link.py      # 短链CRUD
│   │   │   │   └── ws.py        # WebSocket处理器
│   │   │   └── deps.py          # 依赖注入（获取当前用户等）
│   ├── core/
│   │   ├── config.py            # 环境变量配置（pydantic-settings）
│   │   ├── database.py          # 异步引擎 + session工厂
│   │   └── redis_client.py      # Redis连接池
│   ├── models/
│   │   └── link.py              # SQLAlchemy ORM模型
│   ├── schemas/
│   │   └── link.py              # Pydantic请求/响应模型
│   ├── services/
│   │   └── link_service.py      # 业务逻辑（生成短码、访问计数）
│   └── main.py                  # 应用入口
├── static/                      # 前端静态资源（Vue构建）
├── nginx/nginx.conf
├── docker-compose.yml
├── Dockerfile
└── requirements.txt

from pydantic_settings import BaseSettings
from pydantic import PostgresDsn, RedisDsn

class Settings(BaseSettings):
    PROJECT_NAME: str = "ShortLink"
    VERSION: str = "1.0.0"
    
    # PostgreSQL 异步连接串
    DATABASE_URL: PostgresDsn = "postgresql+asyncpg://user:pass@localhost:5432/shortlink"
    
    # Redis 连接串（带密码）
    REDIS_URL: RedisDsn = "redis://:password@localhost:6379/0"
    
    # 短码长度与字符集
    SHORT_CODE_LENGTH: int = 6
    ALLOWED_CHARS: str = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
    
    # WebSocket 心跳间隔
    WS_HEARTBEAT: int = 30
    
    class Config:
        env_file = ".env"

settings = Settings()

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import declarative_base

engine = create_async_engine(
    settings.DATABASE_URL,
    pool_size=20,           # 连接池大小
    max_overflow=10,        # 超出pool_size时额外创建的连接数
    pool_timeout=30,        # 获取连接超时
    pool_recycle=3600,      # 1小时后回收连接（避免pg服务端超时）
    echo=False              # 生产环境关闭SQL日志
)

AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

# 用于依赖注入的会话获取
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

from redis.asyncio import Redis, ConnectionPool

redis_pool = ConnectionPool.from_url(
    settings.REDIS_URL,
    max_connections=50,
    decode_responses=True,
    health_check_interval=30
)

async def get_redis() -> Redis:
    return Redis(connection_pool=redis_pool)

# 在 FastAPI 启动事件中预热
@app.on_event("startup")
async def startup():
    redis = await get_redis()
    await redis.ping()  # 验证连接
    await redis.close()

from sqlalchemy import Column, String, Integer, DateTime, BigInteger, Index
from sqlalchemy.sql import func
from app.core.database import Base

class ShortLink(Base):
    __tablename__ = "short_links"
    
    id = Column(BigInteger, primary_key=True, index=True)
    short_code = Column(String(10), unique=True, index=True, nullable=False)
    original_url = Column(String(2048), nullable=False)
    clicks = Column(Integer, default=0)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    expires_at = Column(DateTime(timezone=True), nullable=True)
    
    # 复合索引：过期时间+短码，用于查询未过期链接
    __table_args__ = (
        Index("ix_expires_short", "expires_at", "short_code"),
    )

from pydantic import BaseModel, HttpUrl, validator
from datetime import datetime
import re

class LinkCreate(BaseModel):
    original_url: HttpUrl  # 自动校验URL格式
    custom_code: str | None = None
    expire_days: int | None = None  # 可选过期天数
    
    @validator('custom_code')
    def validate_custom_code(cls, v):
        if v and not re.match(r'^[A-Za-z0-9_\-]{4,20}$', v):
            raise ValueError('自定义码必须为4-20位字母数字或下划线')
        return v

class LinkResponse(BaseModel):
    short_code: str
    original_url: str
    short_url: str
    clicks: int
    created_at: datetime
    expires_at: datetime | None
    
    class Config:
        from_attributes = True  # 支持ORM对象转换

import base64
import hashlib
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update
from app.models.link import ShortLink
from app.schemas.link import LinkCreate
from app.core.redis_client import get_redis
from app.core.config import settings

class LinkService:
    
    @staticmethod
    async def create_short_link(db: AsyncSession, data: LinkCreate) -> ShortLink:
        # 短码生成策略：对原URL做SHA256，截取前6字节后base62编码
        if data.custom_code:
            short_code = data.custom_code
            # 检查是否已存在
            existing = await db.execute(select(ShortLink).where(ShortLink.short_code == short_code))
            if existing.scalar_one_or_none():
                raise ValueError("自定义短码已被占用")
        else:
            short_code = LinkService._generate_short_code(data.original_url)
            # 极小概率碰撞，循环重试（最多3次）
            for _ in range(3):
                existing = await db.execute(select(ShortLink).where(ShortLink.short_code == short_code))
                if not existing.scalar_one_or_none():
                    break
                short_code = LinkService._generate_short_code(data.original_url + str(short_code))
        
        db_link = ShortLink(
            short_code=short_code,
            original_url=str(data.original_url),
            expires_at=datetime.utcnow() + timedelta(days=data.expire_days) if data.expire_days else None
        )
        db.add(db_link)
        await db.commit()
        await db.refresh(db_link)
        
        # 写入缓存（过期时间与数据库一致）
        redis = await get_redis()
        ttl = data.expire_days * 86400 if data.expire_days else 604800  # 默认7天
        await redis.setex(f"link:{short_code}", ttl, str(data.original_url))
        
        return db_link
    
    @staticmethod
    async def get_original_url(db: AsyncSession, short_code: str) -> str | None:
        # 1. 先查缓存（热点数据）
        redis = await get_redis()
        cached = await redis.get(f"link:{short_code}")
        if cached:
            # 异步增加访问计数（但不阻塞返回）
            asyncio.create_task(LinkService._increment_clicks(db, short_code))
            return cached
        
        # 2. 查数据库
        result = await db.execute(
            select(ShortLink).where(
                ShortLink.short_code == short_code,
                (ShortLink.expires_at > datetime.utcnow()) | (ShortLink.expires_at.is_(None))
            )
        )
        link = result.scalar_one_or_none()
        if not link:
            return None
        
        # 3. 回填缓存
        ttl = int((link.expires_at - datetime.utcnow()).total_seconds()) if link.expires_at else 604800
        await redis.setex(f"link:{short_code}", ttl, link.original_url)
        
        # 异步增加计数
        asyncio.create_task(LinkService._increment_clicks(db, short_code))
        return link.original_url
    
    @staticmethod
    async def _increment_clicks(db: AsyncSession, short_code: str):
        # 使用异步更新，避免select + save的竞态
        await db.execute(
            update(ShortLink).where(ShortLink.short_code == short_code).values(clicks=ShortLink.clicks + 1)
        )
        await db.commit()
    
    @staticmethod
    def _generate_short_code(url: str) -> str:
        # 使用SHA256 + base62编码，缩短长度
        digest = hashlib.sha256(url.encode()).digest()
        # 取前6字节转换为整型，再base62
        num = int.from_bytes(digest[:6], 'big')
        alphabet = settings.ALLOWED_CHARS
        code = ""
        for _ in range(settings.SHORT_CODE_LENGTH):
            code += alphabet[num % 62]
            num //= 62
        return code

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.v1.deps import get_db, get_current_user
from app.schemas.link import LinkCreate, LinkResponse
from app.services.link_service import LinkService

router = APIRouter()

@router.post("/shorten", response_model=LinkResponse, status_code=201)
async def create_short_link(
    data: LinkCreate,
    db: AsyncSession = Depends(get_db),
    # 假设有用户认证，此处可注入当前用户
):
    try:
        link = await LinkService.create_short_link(db, data)
        # 构建短链接完整地址（从请求头获取host）
        short_url = f"https://{request.headers['host']}/{link.short_code}"
        return LinkResponse(
            short_code=link.short_code,
            original_url=link.original_url,
            short_url=short_url,
            clicks=link.clicks,
            created_at=link.created_at,
            expires_at=link.expires_at
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

@router.get("/{short_code}", response_class=RedirectResponse)
async def redirect_short_link(
    short_code: str,
    db: AsyncSession = Depends(get_db)
):
    original_url = await LinkService.get_original_url(db, short_code)
    if not original_url:
        raise HTTPException(status_code=404, detail="短链已过期或不存在")
    return RedirectResponse(url=original_url)

from fastapi import WebSocket, WebSocketDisconnect
from app.core.redis_client import get_redis
import json

class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []
    
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
        await self._broadcast_count()
    
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
        asyncio.create_task(self._broadcast_count())
    
    async def _broadcast_count(self):
        count = len(self.active_connections)
        message = json.dumps({"type": "online_count", "count": count})
        # 通过Redis发布，使多实例同步
        redis = await get_redis()
        await redis.publish("ws:online", message)
        for conn in self.active_connections:
            try:
                await conn.send_text(message)
            except:
                pass

manager = ConnectionManager()

@router.websocket("/ws/online")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            # 接收心跳（或客户端消息）
            data = await websocket.receive_text()
            # 可处理自定义消息
    except WebSocketDisconnect:
        manager.disconnect(websocket)

from fastapi import Request
import time
import logging

logger = logging.getLogger("uvicorn.access")

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    process_time = (time.perf_counter() - start) * 1000
    if process_time > 200:  # 超过200ms警告
        logger.warning(f"Slow request: {request.url.path} took {process_time:.2f}ms")
    response.headers["X-Process-Time"] = f"{process_time:.2f}ms"
    return response

from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail, "code": exc.status_code}
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "code": 422}
    )

import multiprocessing

bind = "0.0.0.0:8000"
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
max_requests = 1000
max_requests_jitter = 100
preload_app = True  # 预加载应用减少内存

gunicorn -k uvicorn.workers.UvicornWorker app.main:app

upstream backend {
    least_conn;  # 最少连接负载均衡
    server web:8000;
}

server {
    listen 80;
    client_max_body_size 10M;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header Host $host;

    # 静态文件（由Vue构建）直接由Nginx返回，缓存1年
    location /static/ {
        alias /app/static/;
        expires 1y;
        add_header Cache-Control "public, immutable";
    }

    # WebSocket代理升级
    location /ws/ {
        proxy_pass http://backend;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_read_timeout 60s;
    }

    # API和重定向
    location / {
        proxy_pass http://backend;
        proxy_cache_key "$scheme$request_method$host$request_uri";
        proxy_cache_valid 200 302 60s;  # 缓存重定向响应
    }
}

version: '3.8'
services:
  postgres:
    image: postgres:15-alpine
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: shortlink
    volumes:
      - pg_data:/var/lib/postgresql/data
    command: -c 'max_connections=100' -c 'shared_buffers=256MB'
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user"]
      interval: 10s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    command: redis-server --requirepass password --maxmemory 512mb --maxmemory-policy allkeys-lru
    volumes:
      - redis_data:/data

  web:
    build: .
    environment:
      DATABASE_URL: postgresql+asyncpg://user:pass@postgres:5432/shortlink
      REDIS_URL: redis://:password@redis:6379/0
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_started
    volumes:
      - ./static:/app/static
    command: gunicorn -c gunicorn.conf.py app.main:app
    restart: unless-stopped

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/conf.d/default.conf:ro
      - ./static:/app/static:ro
    depends_on:
      - web