1.什么是 Uvicorn？ #

Uvicorn 是一个轻量级、极速的 ASGI（Asynchronous Server Gateway Interface）服务器，专门用于运行异步 Python web 应用。它的名字来源于 "UV" (使用 uvloop) + "icorn" (源自 unicorn)。

2.核心特性 #

1. 极高性能：基于 uvloop 和 httptools，性能接近 Go 和 Node.js
2. ASGI 兼容：支持 FastAPI、Starlette、Django 3+ 等异步框架
3. 热重载：开发时支持代码修改自动重启
4. 轻量级：依赖少，启动快速
5. 生产就绪：支持多种配置和部署选项

3.安装 Uvicorn #

安装是使用 Uvicorn 的第一步。本节提供了两种安装方式：基本安装和标准安装。标准安装包含了性能优化的依赖包，推荐在生产环境中使用。

3.1 基本安装 #

# 安装 Uvicorn 基础版本
# 这个命令会安装 Uvicorn 的核心功能
pip install uvicorn

3.2 安装标准版本（推荐） #

# 安装 Uvicorn 标准版本，包含性能优化依赖
# 这会安装 uvloop、httptools 等性能优化包
# 推荐在生产环境中使用此版本
pip install uvicorn[standard]

4.基本用法 #

基本用法展示了如何使用 Uvicorn 启动一个简单的 FastAPI 应用。这是学习 Uvicorn 的起点，包含了最常用的启动方式和参数配置。

4.1 最简单的启动方式 #

假设你有一个 FastAPI 应用在 main.py 中：

# main.py - 创建一个简单的 FastAPI 应用
# 导入 FastAPI 框架
from fastapi import FastAPI

# 创建 FastAPI 应用实例
app = FastAPI()

# 定义根路径的 GET 请求处理函数
# 当用户访问 "/" 时，返回 Hello World 消息
@app.get("/")
async def read_root():
    return {"Hello": "World"}

# 定义带路径参数的 GET 请求处理函数
# item_id 是路径参数，q 是查询参数
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

4.2 启动命令： #

# 启动 Uvicorn 服务器
# main:app 表示从 main.py 模块中导入 app 对象
# --reload 启用热重载，开发时代码修改会自动重启服务器
uvicorn main:app --reload

4.3 指定主机和端口 #

# 启动 Uvicorn 服务器并指定配置
# --host 0.0.0.0 表示绑定到所有网络接口
# --port 8000 指定服务器端口为 8000
# --reload 启用热重载功能
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

5.命令行参数详解 #

命令行参数是 Uvicorn 配置的核心。本节详细介绍了各种启动参数的含义和用法，帮助用户根据不同的需求配置服务器。掌握这些参数对于开发和生产环境部署都非常重要。

5.1 常用参数 #

参数	说明	示例
--host	绑定主机	--host 0.0.0.0
--port	绑定端口	--port 8000
--reload	热重载开发模式	--reload
--workers	工作进程数	--workers 4
--log-level	日志级别	--log-level debug
--ssl-keyfile	SSL 密钥文件	--ssl-keyfile key.pem
--ssl-certfile	SSL 证书文件	--ssl-certfile cert.pem

5.2 完整示例 #

# 启动 Uvicorn 服务器，使用完整的生产环境配置
# --host 0.0.0.0 绑定到所有网络接口，允许外部访问
# --port 8000 指定服务器端口为 8000
# --workers 4 启动 4 个工作进程，提高并发处理能力
# --log-level info 设置日志级别
# --timeout-keep-alive 5 设置保持连接超时时间为 5 秒
uvicorn main:app \
  --host 0.0.0.0 \
  --port 8000 \
  --workers 4 \
  --log-level info \
  --timeout-keep-alive 5

6.编程方式使用 Uvicorn #

除了命令行启动，Uvicorn 还支持在 Python 代码中直接启动。这种方式提供了更大的灵活性，可以动态配置参数，集成到现有的 Python 应用中，或者创建自定义的启动逻辑。

6.1 在 Python 代码中启动 #

# main_programmatic.py - 在 Python 代码中启动 Uvicorn
# 导入必要的模块
import uvicorn
from fastapi import FastAPI

# 创建 FastAPI 应用实例
app = FastAPI()

# 定义根路径的 GET 请求处理函数
@app.get("/")
async def root():
    return {"message": "Hello World"}

# 定义健康检查端点
@app.get("/health")
async def health():
    return {"status": "healthy"}

# 主程序入口
if __name__ == "__main__":
    # 使用 uvicorn.run() 启动服务器
    # 第一个参数是应用字符串，格式为 "模块名:应用实例名"
    # host 指定绑定的主机地址
    # port 指定服务器端口
    # reload 启用热重载功能
    # workers 指定工作进程数量
    # log_level 设置日志级别
    uvicorn.run(
        "main_programmatic:app",  # 应用字符串
        host="0.0.0.0",           # 绑定所有网络接口
        port=8000,                 # 端口号
        reload=True,               # 启用热重载
        workers=4,                 # 工作进程数
        log_level="info"           # 日志级别
    )

6.2 配置类方式 #

# main_config_class.py - 使用配置类方式启动 Uvicorn
# 导入必要的模块
import uvicorn
from uvicorn.config import Config
from uvicorn.main import Server
from fastapi import FastAPI

# 创建 FastAPI 应用实例
app = FastAPI()

# 定义示例路由
@app.get("/")
async def root():
    return {"message": "Hello from Config Class"}

@app.get("/config")
async def config_info():
    return {"method": "config_class", "status": "running"}

# 主程序入口
if __name__ == "__main__":
    # 创建 Uvicorn 配置对象
    # app 指定应用字符串
    # host 指定绑定的主机地址
    # port 指定服务器端口
    # reload 启用热重载功能
    # log_level 设置日志级别
    config = Config(
        app="main_config_class:app",  # 应用字符串
        host="0.0.0.0",              # 绑定所有网络接口
        port=8000,                    # 端口号
        reload=True,                  # 启用热重载
        log_level="debug"             # 日志级别
    )

    # 创建服务器实例
    server = Server(config)

    # 启动服务器
    server.run()

7.配置文件方式 #

配置文件方式提供了更灵活和可维护的 Uvicorn 配置管理。通过配置文件，可以将复杂的启动参数集中管理，便于在不同环境间切换配置，也便于团队协作和版本控制。

7.1 创建 uvicorn_config.py #

# uvicorn_config.py - Uvicorn 配置文件
# 导入必要的模块
import multiprocessing

# 工作进程数配置
# 通常设置为 CPU 核心数 * 2 + 1
# 这样可以充分利用多核性能，同时避免过多的进程切换开销
workers = multiprocessing.cpu_count() * 2 + 1

# 绑定地址配置
# 格式为 "主机:端口"
# 0.0.0.0 表示绑定到所有网络接口
bind = "0.0.0.0:8000"

# 工作模式配置
# 使用 Uvicorn 的异步工作进程
worker_class = "uvicorn.workers.UvicornWorker"

# 日志配置
# loglevel 设置日志级别
loglevel = "info"
# accesslog 设置访问日志输出位置，"-" 表示标准输出
accesslog = "-"
# errorlog 设置错误日志输出位置，"-" 表示标准输出
errorlog = "-"

# 超时设置
# timeout 设置工作进程超时时间（秒）
timeout = 30
# keepalive 设置保持连接时间（秒）
keepalive = 2

# 热重载配置
# 生产环境建议设置为 False
# 开发环境可以设置为 True
reload = False

# 其他性能配置
# 最大请求数，达到后工作进程会重启
max_requests = 1000
# 最大请求抖动，避免所有进程同时重启
max_requests_jitter = 100
# 预加载应用，提高启动速度
preload_app = True

使用配置文件启动：

# 使用配置文件启动 Uvicorn
# --config 参数指定配置文件路径
uvicorn main:app --config uvicorn_config.py

7.2 创建 uvicorn.yaml 配置文件 #

# uvicorn.yaml - YAML 格式的 Uvicorn 配置文件
# 服务器基本配置
server:
  # 绑定地址和端口
  host: "0.0.0.0"
  port: 8000

  # 工作进程配置
  workers: 4
  worker_class: "uvicorn.workers.UvicornWorker"

  # 热重载配置
  reload: false

  # 日志配置
  log_level: "info"
  access_log: true

  # 超时配置
  timeout_keep_alive: 5
  timeout_graceful_shutdown: 30

  # SSL 配置（可选）
  ssl_keyfile: null
  ssl_certfile: null

  # 性能配置
  limit_concurrency: 1000
  limit_max_requests: 10000
  backlog: 2048

7.3 创建 uvicorn.json 配置文件 #

{
  "server": {
    "host": "0.0.0.0",
    "port": 8000,
    "workers": 4,
    "worker_class": "uvicorn.workers.UvicornWorker",
    "reload": false,
    "log_level": "info",
    "access_log": true,
    "timeout_keep_alive": 5,
    "timeout_graceful_shutdown": 30,
    "limit_concurrency": 1000,
    "limit_max_requests": 10000,
    "backlog": 2048
  },
  "logging": {
    "version": 1,
    "disable_existing_loggers": false,
    "formatters": {
      "default": {
        "format": "%(levelprefix)s %(asctime)s %(message)s",
        "datefmt": "%Y-%m-%d %H:%M:%S"
      }
    },
    "handlers": {
      "default": {
        "formatter": "default",
        "class": "logging.StreamHandler",
        "stream": "ext://sys.stderr"
      }
    },
    "loggers": {
      "uvicorn": {
        "handlers": ["default"],
        "level": "INFO"
      }
    }
  }
}

8.生产环境部署 #

生产环境部署是 Uvicorn 应用上线的关键步骤。本节介绍了多种部署方式，包括进程管理、服务管理、容器化等，帮助用户选择最适合自己环境的部署方案。生产环境部署需要考虑性能、稳定性、监控和维护等多个方面。

8.1 使用 Gunicorn + Uvicorn（推荐） #

# 安装 Gunicorn 和 Uvicorn 标准版本
# Gunicorn 是一个 Python WSGI HTTP 服务器，可以管理多个工作进程
# Uvicorn 提供 ASGI 支持
pip install gunicorn uvicorn[standard]

# 启动命令
# -w 4 指定 4 个工作进程
# -k uvicorn.workers.UvicornWorker 指定使用 Uvicorn 的工作进程类
# main:app 指定应用入口
gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app

创建 gunicorn.conf.py 配置文件：

# gunicorn.conf.py - Gunicorn 配置文件
# 导入必要的模块
import multiprocessing

# 绑定地址和端口
bind = "0.0.0.0:8000"

# 工作进程数
# 通常设置为 CPU 核心数 * 2 + 1
workers = multiprocessing.cpu_count() * 2 + 1

# 工作进程类
# 使用 Uvicorn 的异步工作进程
worker_class = "uvicorn.workers.UvicornWorker"

# 工作进程超时时间（秒）
timeout = 30

# 保持连接超时时间（秒）
keepalive = 2

# 最大请求数，达到后工作进程会重启
max_requests = 1000

# 最大请求抖动，避免所有进程同时重启
max_requests_jitter = 100

# 预加载应用，提高启动速度
preload_app = True

# 日志配置
loglevel = "info"
accesslog = "-"
errorlog = "-"

# 进程名称
proc_name = "uvicorn_app"

# 用户和组（需要 root 权限）
# user = "www-data"
# group = "www-data"

使用配置文件启动：

# 使用配置文件启动 Gunicorn + Uvicorn
gunicorn -c gunicorn.conf.py main:app

8.2 使用 Supervisor 管理 #

创建 /etc/supervisor/conf.d/uvicorn.conf：

# /etc/supervisor/conf.d/uvicorn.conf - Supervisor 配置文件
# 程序名称
[program:uvicorn]
# 启动命令
# 使用虚拟环境中的 uvicorn
command=/path/to/venv/bin/uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
# 工作目录
directory=/path/to/your/app
# 运行用户
user=www-data
# 自动启动
autostart=true
# 自动重启
autorestart=true
# 重定向错误输出
redirect_stderr=true
# 标准输出日志文件
stdout_logfile=/var/log/uvicorn.log
# 标准错误日志文件
stderr_logfile=/var/log/uvicorn_error.log
# 环境变量
environment=PYTHONPATH="/path/to/your/app"
# 启动重试次数
startretries=3
# 启动重试间隔（秒）
startsecs=10
# 停止信号
stopsignal=TERM
# 停止等待时间（秒）
stopwaitsecs=10
# 进程优先级
priority=1000

Supervisor 管理命令：

# 重新加载 Supervisor 配置
sudo supervisorctl reread

# 更新 Supervisor 配置
sudo supervisorctl update

# 启动 uvicorn 程序
sudo supervisorctl start uvicorn

# 停止 uvicorn 程序
sudo supervisorctl stop uvicorn

# 重启 uvicorn 程序
sudo supervisorctl restart uvicorn

# 查看程序状态
sudo supervisorctl status uvicorn

# 查看程序日志
sudo supervisorctl tail uvicorn

8.3 使用 Systemd #

创建 /etc/systemd/system/uvicorn.service：

# /etc/systemd/system/uvicorn.service - Systemd 服务配置文件
# 服务单元描述
[Unit]
Description=Uvicorn ASGI Server
# 在网络服务启动后启动
After=network.target
# 依赖网络服务
Wants=network.target

# 服务配置
[Service]
# 运行用户
User=www-data
# 运行组
Group=www-data
# 工作目录
WorkingDirectory=/path/to/your/app
# 启动命令
ExecStart=/path/to/venv/bin/uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
# 重启策略
Restart=always
# 重启延迟（秒）
RestartSec=5
# 环境变量
Environment=PYTHONPATH=/path/to/your/app
Environment=PYTHONUNBUFFERED=1
# 标准输出重定向
StandardOutput=journal
# 标准错误重定向
StandardError=journal
# 进程类型
Type=notify
# 通知套接字路径
NotifyAccess=main
# 超时设置
TimeoutStartSec=30
TimeoutStopSec=30

# 安装配置
[Install]
# 多用户模式启动时启动
WantedBy=multi-user.target

Systemd 管理命令：

# 重新加载 Systemd 配置
sudo systemctl daemon-reload

# 启用服务（开机自启）
sudo systemctl enable uvicorn

# 启动服务
sudo systemctl start uvicorn

# 停止服务
sudo systemctl stop uvicorn

# 重启服务
sudo systemctl restart uvicorn

# 查看服务状态
sudo systemctl status uvicorn

# 查看服务日志
sudo journalctl -u uvicorn -f

# 禁用服务（取消开机自启）
sudo systemctl disable uvicorn

9.性能优化配置 #

性能优化是生产环境部署的重要环节。本节介绍了如何通过调整 Uvicorn 的各种参数来提升服务器性能，包括进程管理、连接处理、超时设置等。合理的性能配置可以显著提升应用的并发处理能力和响应速度。

9.1 高性能配置示例 #

# 启动 Uvicorn 服务器，使用高性能配置
# --host 0.0.0.0 绑定到所有网络接口
# --port 8000 指定服务器端口
# --workers 8 启动 8 个工作进程，充分利用多核性能
# --loop uvloop 使用 uvloop 事件循环，提升异步性能
# --http httptools 使用 httptools 解析器，提升 HTTP 解析性能
# --timeout-keep-alive 5 设置保持连接超时时间为 5 秒
# --backlog 2048 设置等待连接队列大小为 2048
# --limit-concurrency 1000 限制并发连接数为 1000
# --limit-max-requests 10000 每个工作进程处理 10000 个请求后重启
uvicorn main:app \
  --host 0.0.0.0 \
  --port 8000 \
  --workers 8 \
  --loop uvloop \
  --http httptools \
  --timeout-keep-alive 5 \
  --backlog 2048 \
  --limit-concurrency 1000 \
  --limit-max-requests 10000

9.2 创建性能优化配置文件 #

# performance_config.py - 性能优化配置文件
# 导入必要的模块
import multiprocessing
import os

# 基础配置
# 绑定地址和端口
bind = "0.0.0.0:8000"

# 工作进程数配置
# 根据 CPU 核心数动态调整
cpu_count = multiprocessing.cpu_count()
workers = min(cpu_count * 2 + 1, 16)  # 最大不超过 16 个进程

# 工作进程类
worker_class = "uvicorn.workers.UvicornWorker"

# 事件循环配置
# 使用 uvloop 提升性能（仅 Linux/macOS）
if os.name != "nt":  # 不是 Windows
    loop = "uvloop"
else:
    loop = "asyncio"

# HTTP 解析器配置
# 使用 httptools 提升解析性能
http = "httptools"

# 超时配置
timeout = 30
timeout_keep_alive = 5
timeout_graceful_shutdown = 30

# 连接限制配置
limit_concurrency = 1000  # 并发连接数限制
limit_max_requests = 10000  # 每个进程最大请求数
backlog = 2048  # 等待连接队列大小

# 进程管理配置
max_requests = 10000
max_requests_jitter = 1000  # 随机抖动，避免同时重启
preload_app = True  # 预加载应用

# 日志配置
loglevel = "info"
accesslog = "-"
errorlog = "-"

# 安全配置
forwarded_allow_ips = "*"  # 允许的代理 IP
proxy_headers = True  # 处理代理头

9.3 系统级性能优化 #

创建 system_optimization.sh 脚本：

#!/bin/bash
# system_optimization.sh - 系统级性能优化脚本

echo "开始系统性能优化..."

# 调整网络参数
echo "调整网络参数..."

# 增加 TCP 连接队列大小
echo "net.core.somaxconn = 65535" >> /etc/sysctl.conf

# 增加 TCP 半连接队列大小
echo "net.ipv4.tcp_max_syn_backlog = 65535" >> /etc/sysctl.conf

# 启用 TCP 快速回收
echo "net.ipv4.tcp_tw_reuse = 1" >> /etc/sysctl.conf

# 调整 TCP 缓冲区大小
echo "net.core.rmem_max = 16777216" >> /etc/sysctl.conf
echo "net.core.wmem_max = 16777216" >> /etc/sysctl.conf

# 调整文件描述符限制
echo "fs.file-max = 65535" >> /etc/sysctl.conf

# 应用系统参数
sysctl -p

# 调整用户限制
echo "调整用户限制..."

# 编辑 limits.conf 文件
cat >> /etc/security/limits.conf << EOF
# Uvicorn 应用用户限制
www-data soft nofile 65535
www-data hard nofile 65535
www-data soft nproc 65535
www-data hard nproc 65535
EOF

echo "系统性能优化完成！"
echo "请重启系统或重新登录以应用用户限制更改。"

9.4 性能监控脚本 #

创建 performance_monitor.py 脚本：

# performance_monitor.py - 性能监控脚本
# 导入必要的模块
import psutil
import time
import requests
import json
from datetime import datetime

class UvicornMonitor:
    """Uvicorn 性能监控类"""

    def __init__(self, app_url="http://localhost:8000"):
        # 应用 URL
        self.app_url = app_url
        # 监控间隔（秒）
        self.interval = 5

    def get_system_stats(self):
        """获取系统统计信息"""
        # CPU 使用率
        cpu_percent = psutil.cpu_percent(interval=1)

        # 内存使用情况
        memory = psutil.virtual_memory()

        # 磁盘使用情况
        disk = psutil.disk_usage('/')

        # 网络 I/O
        network = psutil.net_io_counters()

        return {
            "timestamp": datetime.now().isoformat(),
            "cpu_percent": cpu_percent,
            "memory_percent": memory.percent,
            "memory_used_gb": round(memory.used / (1024**3), 2),
            "memory_total_gb": round(memory.total / (1024**3), 2),
            "disk_percent": disk.percent,
            "disk_free_gb": round(disk.free / (1024**3), 2),
            "network_bytes_sent": network.bytes_sent,
            "network_bytes_recv": network.bytes_recv
        }

    def get_app_health(self):
        """获取应用健康状态"""
        try:
            # 发送健康检查请求
            response = requests.get(f"{self.app_url}/health", timeout=5)
            return {
                "status": "healthy",
                "response_time_ms": round(response.elapsed.total_seconds() * 1000, 2),
                "status_code": response.status_code
            }
        except requests.RequestException as e:
            return {
                "status": "unhealthy",
                "error": str(e),
                "response_time_ms": None,
                "status_code": None
            }

    def monitor(self, duration_minutes=60):
        """开始监控"""
        print(f"开始监控 Uvicorn 应用，持续 {duration_minutes} 分钟...")
        print("=" * 80)

        start_time = time.time()
        end_time = start_time + (duration_minutes * 60)

        while time.time() < end_time:
            # 获取系统统计信息
            system_stats = self.get_system_stats()

            # 获取应用健康状态
            app_health = self.get_app_health()

            # 合并监控数据
            monitor_data = {**system_stats, **app_health}

            # 打印监控信息
            self.print_stats(monitor_data)

            # 等待下次监控
            time.sleep(self.interval)

        print("监控完成！")

    def print_stats(self, stats):
        """打印统计信息"""
        print(f"[{stats['timestamp']}] 系统状态:")
        print(f"  CPU: {stats['cpu_percent']}% | "
              f"内存: {stats['memory_percent']}% ({stats['memory_used_gb']}GB/{stats['memory_total_gb']}GB) | "
              f"磁盘: {stats['disk_percent']}% (剩余: {stats['disk_free_gb']}GB)")
        print(f"  应用状态: {stats['status']} | "
              f"响应时间: {stats['response_time_ms']}ms | "
              f"状态码: {stats['status_code']}")
        print("-" * 80)

if __name__ == "__main__":
    # 创建监控实例
    monitor = UvicornMonitor()

    # 开始监控（监控 10 分钟）
    monitor.monitor(duration_minutes=10)

使用监控脚本：

# 安装依赖
pip install psutil requests

# 运行监控脚本
python performance_monitor.py

10.日志配置 #

日志配置是 Uvicorn 应用运维的重要组成部分。合理的日志配置可以帮助开发者快速定位问题，监控应用运行状态，并满足审计和合规要求。本节介绍了如何自定义日志格式、配置日志处理器，以及如何将日志输出到不同的目标。

10.1 自定义日志格式 #

# custom_logging.py - 自定义日志配置示例
# 导入必要的模块
import uvicorn
import logging
from fastapi import FastAPI

# 创建 FastAPI 应用实例
app = FastAPI()

# 定义示例路由
@app.get("/")
async def root():
    return {"message": "Hello World with Custom Logging"}

@app.get("/test")
async def test():
    # 记录一些测试日志
    app.logger.info("测试路由被访问")
    app.logger.warning("这是一个警告消息")
    return {"status": "tested"}

# 自定义日志配置
# 这个配置字典定义了 Uvicorn 的日志行为
log_config = {
    # 日志配置版本
    "version": 1,
    # 是否禁用现有的日志记录器
    "disable_existing_loggers": False,

    # 日志格式器配置
    "formatters": {
        # 默认格式器
        "default": {
            # 使用 Uvicorn 的默认格式器类
            "()": "uvicorn.logging.DefaultFormatter",
            # 日志格式：级别前缀 + 时间 + 消息
            "fmt": "%(levelprefix)s %(asctime)s %(message)s",
            # 时间格式：年-月-日 时:分:秒
            "datefmt": "%Y-%m-%d %H:%M:%S",
        },
        # 访问日志格式器
        "access": {
            # 使用 Uvicorn 的访问日志格式器类
            "()": "uvicorn.logging.AccessFormatter",
            # 访问日志格式：级别前缀 + 时间 + 请求行 + 状态码
            "fmt": '%(levelprefix)s %(asctime)s - "%(request_line)s" %(status_code)s',
            # 时间格式：年-月-日 时:分:秒
            "datefmt": "%Y-%m-%d %H:%M:%S",
        },
        # 详细格式器（用于调试）
        "detailed": {
            "()": "uvicorn.logging.DefaultFormatter",
            # 详细格式：级别前缀 + 时间 + 模块名 + 行号 + 消息
            "fmt": "%(levelprefix)s %(asctime)s [%(name)s:%(lineno)d] %(message)s",
            "datefmt": "%Y-%m-%d %H:%M:%S",
        },
    },

    # 日志处理器配置
    "handlers": {
        # 默认处理器（错误日志）
        "default": {
            "formatter": "default",
            "class": "logging.StreamHandler",
            "stream": "ext://sys.stderr",
        },
        # 访问日志处理器
        "access": {
            "formatter": "access",
            "class": "logging.StreamHandler",
            "stream": "ext://sys.stdout",
        },
        # 文件处理器（可选）
        "file": {
            "formatter": "detailed",
            "class": "logging.FileHandler",
            "filename": "uvicorn.log",
            "mode": "a",
            "encoding": "utf-8",
        },
    },

    # 日志记录器配置
    "loggers": {
        # Uvicorn 主日志记录器
        "uvicorn": {
            "handlers": ["default"],
            "level": "INFO",
            "propagate": False,
        },
        # Uvicorn 错误日志记录器
        "uvicorn.error": {
            "level": "INFO",
            "propagate": False,
        },
        # Uvicorn 访问日志记录器
        "uvicorn.access": {
            "handlers": ["access"],
            "level": "INFO",
            "propagate": False,
        },
        # 应用日志记录器
        "app": {
            "handlers": ["default", "file"],
            "level": "DEBUG",
            "propagate": False,
        },
    },
}

# 主程序入口
if __name__ == "__main__":
    # 使用自定义日志配置启动 Uvicorn
    # log_config 参数指定自定义的日志配置
    uvicorn.run(
        "custom_logging:app",  # 应用字符串
        host="0.0.0.0",        # 绑定所有网络接口
        port=8000,              # 端口号
        log_config=log_config,  # 自定义日志配置
        reload=True             # 启用热重载
    )

10.2 结构化日志配置 #

# structured_logging.py - 结构化日志配置示例
# 导入必要的模块
import uvicorn
import logging
import json
from datetime import datetime
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

# 创建 FastAPI 应用实例
app = FastAPI()

# 结构化日志格式器
class StructuredFormatter(logging.Formatter):
    """结构化日志格式器，输出 JSON 格式的日志"""

    def format(self, record):
        # 创建结构化日志记录
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }

        # 添加异常信息（如果有）
        if record.exc_info:
            log_entry["exception"] = self.formatException(record.exc_info)

        # 添加额外字段（如果有）
        if hasattr(record, "extra_fields"):
            log_entry.update(record.extra_fields)

        return json.dumps(log_entry, ensure_ascii=False)

# 中间件：记录请求日志
@app.middleware("http")
async def log_requests(request: Request, call_next):
    # 记录请求开始
    start_time = datetime.utcnow()

    # 记录请求信息
    app.logger.info("请求开始", extra={
        "extra_fields": {
            "method": request.method,
            "url": str(request.url),
            "client_ip": request.client.host,
            "user_agent": request.headers.get("user-agent", ""),
        }
    })

    # 处理请求
    response = await call_next(request)

    # 计算处理时间
    process_time = (datetime.utcnow() - start_time).total_seconds()

    # 记录响应信息
    app.logger.info("请求完成", extra={
        "extra_fields": {
            "method": request.method,
            "url": str(request.url),
            "status_code": response.status_code,
            "process_time": round(process_time, 4),
        }
    })

    return response

# 定义路由
@app.get("/")
async def root():
    app.logger.info("根路径被访问")
    return {"message": "Hello Structured Logging"}

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    app.logger.info(f"获取商品 {item_id}", extra={
        "extra_fields": {
            "item_id": item_id,
            "operation": "read",
        }
    })
    return {"item_id": item_id, "name": f"商品 {item_id}"}

# 结构化日志配置
structured_log_config = {
    "version": 1,
    "disable_existing_loggers": False,

    "formatters": {
        "structured": {
            "()": StructuredFormatter,
        },
    },

    "handlers": {
        "structured": {
            "formatter": "structured",
            "class": "logging.StreamHandler",
            "stream": "ext://sys.stdout",
        },
    },

    "loggers": {
        "": {  # 根日志记录器
            "handlers": ["structured"],
            "level": "INFO",
        },
    },
}

# 主程序入口
if __name__ == "__main__":
    # 使用结构化日志配置启动 Uvicorn
    uvicorn.run(
        "structured_logging:app",
        host="0.0.0.0",
        port=8000,
        log_config=structured_log_config,
        reload=True
    )

10.3 日志轮转配置 #

# log_rotation.py - 日志轮转配置示例
# 导入必要的模块
import uvicorn
import logging
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
from fastapi import FastAPI

# 创建 FastAPI 应用实例
app = FastAPI()

# 定义示例路由
@app.get("/")
async def root():
    app.logger.info("根路径被访问")
    return {"message": "Hello Log Rotation"}

# 日志轮转配置
rotation_log_config = {
    "version": 1,
    "disable_existing_loggers": False,

    "formatters": {
        "detailed": {
            "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
            "datefmt": "%Y-%m-%d %H:%M:%S",
        },
    },

    "handlers": {
        # 按大小轮转的日志文件
        "rotating_file": {
            "class": "logging.handlers.RotatingFileHandler",
            "filename": "uvicorn.log",
            "maxBytes": 10 * 1024 * 1024,  # 10MB
            "backupCount": 5,               # 保留 5 个备份文件
            "formatter": "detailed",
            "encoding": "utf-8",
        },
        # 按时间轮转的日志文件
        "timed_file": {
            "class": "logging.handlers.TimedRotatingFileHandler",
            "filename": "uvicorn_daily.log",
            "when": "midnight",             # 每天午夜轮转
            "interval": 1,                  # 间隔 1 天
            "backupCount": 30,              # 保留 30 个备份文件
            "formatter": "detailed",
            "encoding": "utf-8",
        },
        # 控制台输出
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "detailed",
        },
    },

    "loggers": {
        "uvicorn": {
            "handlers": ["rotating_file", "timed_file", "console"],
            "level": "INFO",
            "propagate": False,
        },
        "uvicorn.error": {
            "handlers": ["rotating_file", "timed_file", "console"],
            "level": "INFO",
            "propagate": False,
        },
        "uvicorn.access": {
            "handlers": ["rotating_file", "timed_file", "console"],
            "level": "INFO",
            "propagate": False,
        },
    },
}

# 主程序入口
if __name__ == "__main__":
    # 使用日志轮转配置启动 Uvicorn
    uvicorn.run(
        "log_rotation:app",
        host="0.0.0.0",
        port=8000,
        log_config=rotation_log_config,
        reload=True
    )

11.SSL/TLS 配置 #

SSL/TLS 配置是生产环境部署中保障数据传输安全的重要环节。本节介绍了如何在 Uvicorn 中启用 HTTPS，包括证书配置、安全参数设置等。正确的 SSL/TLS 配置可以保护用户数据安全，提升应用的可信度。

11.1 启用 HTTPS #

# 启动 Uvicorn 服务器，启用 HTTPS
# --host 0.0.0.0 绑定到所有网络接口
# --port 8443 指定 HTTPS 端口（标准 HTTPS 端口是 443）
# --ssl-keyfile 指定私钥文件路径
# --ssl-certfile 指定证书文件路径
# --ssl-ca-certs 指定 CA 证书包路径（可选）
uvicorn main:app \
  --host 0.0.0.0 \
  --port 8443 \
  --ssl-keyfile /path/to/private.key \
  --ssl-certfile /path/to/certificate.crt \
  --ssl-ca-certs /path/to/ca_bundle.crt

11.2 创建自签名证书 #

创建 generate_self_signed_cert.sh 脚本：

#!/bin/bash
# generate_self_signed_cert.sh - 生成自签名 SSL 证书

echo "开始生成自签名 SSL 证书..."

# 创建证书目录
mkdir -p ssl_certs
cd ssl_certs

# 生成私钥
echo "生成私钥..."
openssl genrsa -out private.key 2048

# 生成证书签名请求（CSR）
echo "生成证书签名请求..."
openssl req -new -key private.key -out certificate.csr -subj "/C=CN/ST=Beijing/L=Beijing/O=MyCompany/CN=localhost"

# 生成自签名证书
echo "生成自签名证书..."
openssl x509 -req -in certificate.csr -signkey private.key -out certificate.crt -days 365

# 设置权限
chmod 600 private.key
chmod 644 certificate.crt

echo "自签名证书生成完成！"
echo "私钥文件: ssl_certs/private.key"
echo "证书文件: ssl_certs/certificate.crt"
echo ""
echo "注意：自签名证书仅适用于开发和测试环境，生产环境请使用受信任的 CA 颁发的证书。"

11.3 SSL 配置示例 #

# ssl_config.py - SSL 配置示例
# 导入必要的模块
import uvicorn
from fastapi import FastAPI
from fastapi.responses import JSONResponse

# 创建 FastAPI 应用实例
app = FastAPI()

# 定义示例路由
@app.get("/")
async def root():
    return {"message": "Hello HTTPS World"}

@app.get("/secure")
async def secure_endpoint():
    return {"message": "这是一个安全的 HTTPS 端点"}

# SSL 配置
ssl_config = {
    "ssl_keyfile": "ssl_certs/private.key",      # 私钥文件路径
    "ssl_certfile": "ssl_certs/certificate.crt", # 证书文件路径
    "ssl_ca_certs": None,                        # CA 证书包路径（可选）
    "ssl_version": "TLSv1_2",                    # TLS 版本
    "ssl_ciphers": "ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256",  # 加密套件
}

# 主程序入口
if __name__ == "__main__":
    # 使用 SSL 配置启动 Uvicorn
    uvicorn.run(
        "ssl_config:app",
        host="0.0.0.0",
        port=8443,
        ssl_keyfile=ssl_config["ssl_keyfile"],
        ssl_certfile=ssl_config["ssl_certfile"],
        ssl_ca_certs=ssl_config["ssl_ca_certs"],
        reload=True
    )

11.4 反向代理 SSL 配置 #

创建 nginx_ssl.conf 配置文件：

# nginx_ssl.conf - Nginx SSL 反向代理配置
server {
    # 监听 HTTPS 端口
    listen 443 ssl http2;
    server_name your-domain.com;

    # SSL 证书配置
    ssl_certificate /path/to/certificate.crt;
    ssl_certificate_key /path/to/private.key;

    # SSL 安全配置
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256;
    ssl_prefer_server_ciphers off;
    ssl_session_cache shared:SSL:10m;
    ssl_session_timeout 10m;

    # 安全头
    add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
    add_header X-Frame-Options DENY always;
    add_header X-Content-Type-Options nosniff always;
    add_header X-XSS-Protection "1; mode=block" always;

    # 反向代理到 Uvicorn
    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # WebSocket 支持
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}

# HTTP 重定向到 HTTPS
server {
    listen 80;
    server_name your-domain.com;
    return 301 https://$server_name$request_uri;
}

12.常见问题解决 #

在实际使用 Uvicorn 的过程中，经常会遇到各种问题。本节总结了最常见的问题及其解决方案，包括端口占用、权限问题、性能调优等。掌握这些解决方案可以帮助用户快速排除故障，提高应用稳定性。

12.1 端口被占用 #

Windows 系统： #

# 查找占用端口的进程
# netstat 显示网络连接状态
# -ano 显示所有连接和监听端口，以及对应的进程 ID
netstat -ano | findstr :8000

# 杀死进程（替换 PID 为实际的进程 ID）
# taskkill 用于终止进程
# /F 强制终止
# /PID 指定进程 ID
taskkill /F /PID <PID>

Linux/macOS 系统： #

# 查找占用端口的进程
# lsof 列出打开的文件
# -i :8000 显示端口 8000 的连接
lsof -i :8000

# 或者使用 netstat
netstat -tulpn | grep :8000

# 杀死进程
kill -9 <PID>

Python 脚本解决方案： #

# port_checker.py - 端口检查和释放脚本
import socket
import psutil
import subprocess
import platform
import sys

def check_port(port):
    """检查端口是否被占用"""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        try:
            s.bind(('localhost', port))
            return False  # 端口可用
        except OSError:
            return True   # 端口被占用

def find_process_using_port(port):
    """查找占用端口的进程"""
    for proc in psutil.process_iter(['pid', 'name', 'connections']):
        try:
            connections = proc.info['connections']
            if connections:
                for conn in connections:
                    if conn.laddr.port == port:
                        return proc.info
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    return None

def kill_process(pid):
    """杀死指定进程"""
    system = platform.system().lower()

    try:
        if system == "windows":
            subprocess.run(['taskkill', '/F', '/PID', str(pid)], check=True)
        else:
            subprocess.run(['kill', '-9', str(pid)], check=True)
        print(f"进程 {pid} 已被终止")
        return True
    except subprocess.CalledProcessError as e:
        print(f"无法终止进程 {pid}: {e}")
        return False

def main():
    port = 8000

    print(f"检查端口 {port} 的状态...")

    if check_port(port):
        print(f"端口 {port} 被占用")

        # 查找占用端口的进程
        proc_info = find_process_using_port(port)
        if proc_info:
            print(f"占用进程: PID={proc_info['pid']}, 名称={proc_info['name']}")

            # 询问是否终止进程
            response = input("是否终止此进程？(y/n): ")
            if response.lower() == 'y':
                if kill_process(proc_info['pid']):
                    print(f"端口 {port} 现在应该可用了")
                else:
                    print("无法释放端口")
            else:
                print("端口仍被占用")
        else:
            print("无法找到占用端口的进程信息")
    else:
        print(f"端口 {port} 可用")

if __name__ == "__main__":
    main()

12.2 权限问题 #

Linux 系统端口权限： #

# 如果无法绑定 80 端口，使用端口转发
# iptables 是 Linux 防火墙工具
# -t nat 指定 NAT 表
# -A PREROUTING 添加 PREROUTING 链规则
# -p tcp 指定 TCP 协议
# --dport 80 指定目标端口为 80
# -j REDIRECT 执行重定向动作
# --to-port 8000 重定向到端口 8000
sudo iptables -t nat -A PREROUTING -p tcp --dport 80 -j REDIRECT --to-port 8000

# 保存 iptables 规则
sudo iptables-save > /etc/iptables/rules.v4

# 或者使用 ufw（Ubuntu 防火墙）
sudo ufw allow 80
sudo ufw allow 8000

用户权限配置： #

# 创建专用用户运行 Uvicorn
sudo useradd -r -s /bin/false uvicorn_user

# 设置应用目录权限
sudo chown -R uvicorn_user:uvicorn_user /path/to/your/app
sudo chmod -R 755 /path/to/your/app

# 在 systemd 服务文件中使用此用户
# User=uvicorn_user
# Group=uvicorn_user

12.3 性能调优 #

系统级优化脚本： #

#!/bin/bash
# system_tuning.sh - 系统性能调优脚本

echo "开始系统性能调优..."

# 调整网络参数
echo "调整网络参数..."

# 增加 TCP 连接队列大小
echo "net.core.somaxconn = 65535" >> /etc/sysctl.conf

# 增加 TCP 半连接队列大小
echo "net.ipv4.tcp_max_syn_backlog = 65535" >> /etc/sysctl.conf

# 启用 TCP 快速回收
echo "net.ipv4.tcp_tw_reuse = 1" >> /etc/sysctl.conf

# 调整 TCP 缓冲区大小
echo "net.core.rmem_max = 16777216" >> /etc/sysctl.conf
echo "net.core.wmem_max = 16777216" >> /etc/sysctl.conf

# 调整文件描述符限制
echo "fs.file-max = 65535" >> /etc/sysctl.conf

# 应用系统参数
sysctl -p

# 调整用户限制
echo "调整用户限制..."

# 编辑 limits.conf 文件
cat >> /etc/security/limits.conf << EOF
# Uvicorn 应用用户限制
uvicorn_user soft nofile 65535
uvicorn_user hard nofile 65535
uvicorn_user soft nproc 65535
uvicorn_user hard nproc 65535
EOF

echo "系统性能调优完成！"
echo "请重启系统或重新登录以应用用户限制更改。"

Python 性能监控脚本： #

# performance_debug.py - 性能调试脚本
import uvicorn
import asyncio
import time
import psutil
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware

# 创建 FastAPI 应用实例
app = FastAPI()

# 添加 CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 性能监控中间件
@app.middleware("http")
async def performance_monitor(request: Request, call_next):
    # 记录请求开始时间
    start_time = time.time()

    # 记录请求前的系统状态
    process = psutil.Process()
    cpu_before = process.cpu_percent()
    memory_before = process.memory_info().rss / 1024 / 1024  # MB

    # 处理请求
    response = await call_next(request)

    # 记录请求后的系统状态
    cpu_after = process.cpu_percent()
    memory_after = process.memory_info().rss / 1024 / 1024  # MB

    # 计算处理时间
    process_time = time.time() - start_time

    # 记录性能指标
    print(f"请求: {request.method} {request.url}")
    print(f"处理时间: {process_time:.4f} 秒")
    print(f"CPU 使用: {cpu_before:.1f}% -> {cpu_after:.1f}%")
    print(f"内存使用: {memory_before:.1f}MB -> {memory_after:.1f}MB")
    print("-" * 50)

    return response

# 定义测试路由
@app.get("/")
async def root():
    return {"message": "Hello Performance Monitor"}

@app.get("/slow")
async def slow_endpoint():
    # 模拟慢操作
    await asyncio.sleep(1)
    return {"message": "Slow operation completed"}

@app.get("/memory")
async def memory_intensive():
    # 模拟内存密集型操作
    large_list = [i for i in range(100000)]
    return {"message": f"Created list with {len(large_list)} items"}

@app.get("/cpu")
async def cpu_intensive():
    # 模拟 CPU 密集型操作
    result = sum(i * i for i in range(10000))
    return {"message": f"CPU calculation result: {result}"}

# 健康检查端点
@app.get("/health")
async def health_check():
    process = psutil.Process()
    return {
        "status": "healthy",
        "cpu_percent": process.cpu_percent(),
        "memory_mb": round(process.memory_info().rss / 1024 / 1024, 2),
        "threads": process.num_threads(),
        "open_files": len(process.open_files()),
        "connections": len(process.connections())
    }

if __name__ == "__main__":
    # 启动性能监控应用
    uvicorn.run(
        "performance_debug:app",
        host="0.0.0.0",
        port=8000,
        reload=True,
        log_level="info"
    )

12.4 连接问题诊断 #

网络连接测试脚本： #

# connection_test.py - 网络连接测试脚本
import socket
import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

def test_tcp_connection(host, port, timeout=5):
    """测试 TCP 连接"""
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(timeout)
            result = s.connect_ex((host, port))
            return result == 0, f"TCP 连接 {'成功' if result == 0 else '失败'}"
    except Exception as e:
        return False, f"TCP 连接错误: {e}"

def test_http_connection(url, timeout=10):
    """测试 HTTP 连接"""
    try:
        start_time = time.time()
        response = requests.get(url, timeout=timeout)
        response_time = time.time() - start_time

        return True, {
            "status_code": response.status_code,
            "response_time": round(response_time, 3),
            "content_length": len(response.content)
        }
    except requests.RequestException as e:
        return False, str(e)

def test_connection_pool(host, port, connections=100, timeout=5):
    """测试连接池性能"""
    results = []

    def create_connection():
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(timeout)
                s.connect((host, port))
                return True
        except Exception:
            return False

    # 使用线程池创建多个连接
    with ThreadPoolExecutor(max_workers=connections) as executor:
        futures = [executor.submit(create_connection) for _ in range(connections)]

        for future in as_completed(futures):
            results.append(future.result())

    successful = sum(results)
    failed = len(results) - successful

    return {
        "total_connections": len(results),
        "successful": successful,
        "failed": failed,
        "success_rate": round(successful / len(results) * 100, 2)
    }

def main():
    host = "localhost"
    port = 8000
    url = f"http://{host}:{port}"

    print("开始网络连接测试...")
    print("=" * 50)

    # 测试 TCP 连接
    print("1. 测试 TCP 连接...")
    tcp_success, tcp_message = test_tcp_connection(host, port)
    print(f"   {tcp_message}")

    # 测试 HTTP 连接
    print("\n2. 测试 HTTP 连接...")
    if tcp_success:
        http_success, http_result = test_http_connection(url)
        if http_success:
            print(f"   HTTP 连接成功")
            print(f"   状态码: {http_result['status_code']}")
            print(f"   响应时间: {http_result['response_time']} 秒")
            print(f"   内容长度: {http_result['content_length']} 字节")
        else:
            print(f"   HTTP 连接失败: {http_result}")
    else:
        print("   TCP 连接失败，跳过 HTTP 测试")

    # 测试连接池
    print("\n3. 测试连接池性能...")
    pool_result = test_connection_pool(host, port, connections=50)
    print(f"   总连接数: {pool_result['total_connections']}")
    print(f"   成功连接: {pool_result['successful']}")
    print(f"   失败连接: {pool_result['failed']}")
    print(f"   成功率: {pool_result['success_rate']}%")

    print("\n连接测试完成！")

if __name__ == "__main__":
    main()

13.监控和健康检查 #

监控和健康检查是生产环境运维的重要组成部分。通过建立完善的监控体系，可以及时发现和解决潜在问题，确保应用的稳定性和可用性。本节介绍了如何实现健康检查端点、性能监控、日志监控等功能，帮助运维人员更好地管理 Uvicorn 应用。

13.1 添加健康检查端点 #

# health_check.py - 健康检查端点示例
# 导入必要的模块
import uvicorn
import psutil
import time
import asyncio
from datetime import datetime
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel

# 创建 FastAPI 应用实例
app = FastAPI(title="健康检查应用", version="1.0.0")

# 健康状态数据模型
class HealthStatus(BaseModel):
    status: str
    timestamp: str
    uptime: float
    version: str
    environment: str

class SystemInfo(BaseModel):
    cpu_percent: float
    memory_percent: float
    memory_used_gb: float
    memory_total_gb: float
    disk_percent: float
    disk_free_gb: float

class DetailedHealth(BaseModel):
    status: str
    timestamp: str
    uptime: float
    version: str
    environment: str
    system: SystemInfo
    services: dict
    checks: dict

# 应用启动时间
start_time = time.time()

# 模拟外部服务检查
async def check_database():
    """检查数据库连接"""
    try:
        # 模拟数据库连接检查
        await asyncio.sleep(0.1)
        return {"status": "healthy", "response_time": 0.1}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

async def check_cache():
    """检查缓存服务"""
    try:
        # 模拟缓存服务检查
        await asyncio.sleep(0.05)
        return {"status": "healthy", "response_time": 0.05}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

async def check_external_api():
    """检查外部 API"""
    try:
        # 模拟外部 API 检查
        await asyncio.sleep(0.2)
        return {"status": "healthy", "response_time": 0.2}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

# 基础健康检查端点
@app.get("/health", response_model=HealthStatus)
async def health_check():
    """基础健康检查端点"""
    # 计算应用运行时间
    uptime = time.time() - start_time

    return HealthStatus(
        status="healthy",
        timestamp=datetime.now().isoformat(),
        uptime=round(uptime, 2),
        version="1.0.0",
        environment="production"
    )

# 详细健康检查端点
@app.get("/health/detailed", response_model=DetailedHealth)
async def detailed_health_check():
    """详细健康检查端点"""
    # 计算应用运行时间
    uptime = time.time() - start_time

    # 获取系统信息
    cpu_percent = psutil.cpu_percent(interval=1)
    memory = psutil.virtual_memory()
    disk = psutil.disk_usage('/')

    system_info = SystemInfo(
        cpu_percent=cpu_percent,
        memory_percent=memory.percent,
        memory_used_gb=round(memory.used / (1024**3), 2),
        memory_total_gb=round(memory.total / (1024**3), 2),
        disk_percent=disk.percent,
        disk_free_gb=round(disk.free / (1024**3), 2)
    )

    # 检查各个服务
    db_check = await check_database()
    cache_check = await check_cache()
    api_check = await check_external_api()

    services = {
        "database": db_check,
        "cache": cache_check,
        "external_api": api_check
    }

    # 执行健康检查
    checks = {
        "system_resources": "healthy" if cpu_percent < 80 and memory.percent < 80 else "warning",
        "disk_space": "healthy" if disk.percent < 90 else "warning",
        "response_time": "healthy" if uptime > 0 else "error"
    }

    # 确定整体状态
    overall_status = "healthy"
    if any(check["status"] == "unhealthy" for check in services.values()):
        overall_status = "unhealthy"
    elif any(status == "warning" for status in checks.values()):
        overall_status = "warning"

    return DetailedHealth(
        status=overall_status,
        timestamp=datetime.now().isoformat(),
        uptime=round(uptime, 2),
        version="1.0.0",
        environment="production",
        system=system_info,
        services=services,
        checks=checks
    )

# 就绪检查端点
@app.get("/ready")
async def readiness_check():
    """就绪检查端点，用于 Kubernetes 就绪探针"""
    try:
        # 检查关键服务是否就绪
        db_check = await check_database()
        cache_check = await check_cache()

        if db_check["status"] == "healthy" and cache_check["status"] == "healthy":
            return {"status": "ready", "timestamp": datetime.now().isoformat()}
        else:
            raise HTTPException(status_code=503, detail="服务未就绪")
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"就绪检查失败: {str(e)}")

# 存活检查端点
@app.get("/live")
async def liveness_check():
    """存活检查端点，用于 Kubernetes 存活探针"""
    # 简单的存活检查，只要应用能响应就认为存活
    return {"status": "alive", "timestamp": datetime.now().isoformat()}

# 指标端点
@app.get("/metrics")
async def metrics():
    """Prometheus 格式的指标端点"""
    # 获取系统指标
    cpu_percent = psutil.cpu_percent(interval=1)
    memory = psutil.virtual_memory()
    disk = psutil.disk_usage('/')

    # 计算应用指标
    uptime = time.time() - start_time

    # 返回 Prometheus 格式的指标
    metrics_data = f"""# HELP app_uptime_seconds 应用运行时间（秒）
# TYPE app_uptime_seconds gauge
app_uptime_seconds {uptime}

# HELP system_cpu_percent 系统 CPU 使用率
# TYPE system_cpu_percent gauge
system_cpu_percent {cpu_percent}

# HELP system_memory_percent 系统内存使用率
# TYPE system_memory_percent gauge
system_memory_percent {memory.percent}

# HELP system_disk_percent 系统磁盘使用率
# TYPE system_disk_percent gauge
system_disk_percent {disk.percent}

# HELP app_requests_total 总请求数
# TYPE app_requests_total counter
app_requests_total 0
"""

    return Response(content=metrics_data, media_type="text/plain")

# 主程序入口
if __name__ == "__main__":
    # 启动健康检查应用
    uvicorn.run(
        "health_check:app",
        host="0.0.0.0",
        port=8000,
        reload=True,
        log_level="info"
    )

13.2 健康检查脚本 #

创建 health_monitor.py 脚本：

# health_monitor.py - 健康检查监控脚本
import requests
import time
import json
from datetime import datetime
import argparse
import sys

class HealthMonitor:
    """健康检查监控类"""

    def __init__(self, base_url, interval=30):
        # 基础 URL
        self.base_url = base_url.rstrip('/')
        # 检查间隔（秒）
        self.interval = interval
        # 检查历史
        self.history = []

    def check_health(self, endpoint="/health"):
        """检查指定端点的健康状态"""
        try:
            start_time = time.time()
            response = requests.get(f"{self.base_url}{endpoint}", timeout=10)
            response_time = time.time() - start_time

            if response.status_code == 200:
                data = response.json()
                return {
                    "endpoint": endpoint,
                    "status": "healthy",
                    "response_time": round(response_time, 3),
                    "data": data,
                    "timestamp": datetime.now().isoformat()
                }
            else:
                return {
                    "endpoint": endpoint,
                    "status": "unhealthy",
                    "response_time": round(response_time, 3),
                    "error": f"HTTP {response.status_code}",
                    "timestamp": datetime.now().isoformat()
                }
        except requests.RequestException as e:
            return {
                "endpoint": endpoint,
                "status": "error",
                "response_time": None,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }

    def run_health_checks(self):
        """运行所有健康检查"""
        endpoints = [
            "/health",
            "/health/detailed",
            "/ready",
            "/live"
        ]

        results = []
        for endpoint in endpoints:
            result = self.check_health(endpoint)
            results.append(result)
            self.history.append(result)

            # 保持历史记录在合理范围内
            if len(self.history) > 1000:
                self.history = self.history[-1000:]

        return results

    def print_results(self, results):
        """打印检查结果"""
        print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 健康检查结果:")
        print("=" * 80)

        for result in results:
            status_icon = "" if result["status"] == "healthy" else ""
            print(f"{status_icon} {result['endpoint']}: {result['status']}")

            if result["status"] == "healthy":
                print(f"   响应时间: {result['response_time']}s")
                if "data" in result and "uptime" in result["data"]:
                    print(f"   运行时间: {result['data']['uptime']}s")
            else:
                print(f"   错误: {result.get('error', 'Unknown error')}")

        print("-" * 80)

    def generate_report(self, results):
        """生成健康检查报告"""
        total_checks = len(results)
        healthy_checks = sum(1 for r in results if r["status"] == "healthy")
        unhealthy_checks = total_checks - healthy_checks

        report = {
            "timestamp": datetime.now().isoformat(),
            "summary": {
                "total_checks": total_checks,
                "healthy_checks": healthy_checks,
                "unhealthy_checks": unhealthy_checks,
                "health_percentage": round(healthy_checks / total_checks * 100, 2)
            },
            "details": results
        }

        return report

    def save_report(self, report, filename=None):
        """保存报告到文件"""
        if filename is None:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"health_report_{timestamp}.json"

        try:
            with open(filename, 'w', encoding='utf-8') as f:
                json.dump(report, f, indent=2, ensure_ascii=False)
            print(f"报告已保存到: {filename}")
        except Exception as e:
            print(f"保存报告失败: {e}")

    def monitor(self, duration_minutes=None):
        """开始监控"""
        print(f"开始监控应用健康状态: {self.base_url}")
        print(f"检查间隔: {self.interval} 秒")
        if duration_minutes:
            print(f"监控时长: {duration_minutes} 分钟")
        print("=" * 80)

        start_time = time.time()
        end_time = start_time + (duration_minutes * 60) if duration_minutes else None

        try:
            while True:
                # 运行健康检查
                results = self.run_health_checks()

                # 打印结果
                self.print_results(results)

                # 生成并保存报告
                report = self.generate_report(results)

                # 检查是否需要停止
                if end_time and time.time() >= end_time:
                    print("监控时间结束")
                    break

                # 等待下次检查
                if end_time is None or time.time() + self.interval < end_time:
                    time.sleep(self.interval)

        except KeyboardInterrupt:
            print("\n监控被用户中断")

        # 保存最终报告
        final_report = self.generate_report(self.run_health_checks())
        self.save_report(final_report, "health_report_final.json")
        print("监控完成！")

def main():
    parser = argparse.ArgumentParser(description="Uvicorn 应用健康检查监控")
    parser.add_argument("url", help="应用基础 URL (例如: http://localhost:8000)")
    parser.add_argument("-i", "--interval", type=int, default=30, help="检查间隔（秒，默认: 30）")
    parser.add_argument("-d", "--duration", type=int, help="监控时长（分钟）")

    args = parser.parse_args()

    # 创建监控实例
    monitor = HealthMonitor(args.url, args.interval)

    # 开始监控
    monitor.monitor(args.duration)

if __name__ == "__main__":
    main()

13.3 使用健康检查 #

检查服务状态：

# 基础健康检查
curl http://localhost:8000/health

# 详细健康检查
curl http://localhost:8000/health/detailed

# 就绪检查
curl http://localhost:8000/ready

# 存活检查
curl http://localhost:8000/live

# 指标端点
curl http://localhost:8000/metrics

运行健康监控：

# 安装依赖
pip install requests

# 运行健康监控（监控 10 分钟）
python health_monitor.py http://localhost:8000 -d 10

# 持续监控
python health_monitor.py http://localhost:8000

# 自定义检查间隔
python health_monitor.py http://localhost:8000 -i 60

14.总结 #

本节总结了 Uvicorn 的核心特性和最佳实践，帮助用户快速掌握 Uvicorn 的使用要点。通过本指南的学习，用户应该能够熟练地使用 Uvicorn 部署和管理 Python 异步 Web 应用。

14.1 核心特性总结 #

1. 高性能 ASGI 服务器

   • 基于 uvloop 和 httptools 的极速性能
   • 支持异步 Python Web 框架
   • 接近 Go 和 Node.js 的性能表现

2. 灵活的配置方式

   • 命令行参数配置
   • Python 代码配置
   • 配置文件支持（Python、YAML、JSON）
   • 环境变量配置

3. 生产就绪特性

   • 多进程支持
   • 热重载开发模式
   • SSL/TLS 支持
   • 健康检查和监控
   • 日志配置和轮转

4. 部署和管理

   • Gunicorn + Uvicorn 组合
   • Supervisor 进程管理
   • Systemd 服务管理
   • Docker 容器化支持

14.2 最佳实践建议 #

1. 开发环境

   • 使用 --reload 参数启用热重载
   • 设置合适的日志级别（debug/info）
   • 使用单进程模式进行调试

2. 生产环境

   • 使用多进程模式（workers = CPU核心数 * 2 + 1）
   • 启用 SSL/TLS 加密
   • 配置反向代理（Nginx）
   • 设置健康检查和监控
   • 使用进程管理器（Supervisor/Systemd）

3. 性能优化

   • 使用 uvloop 事件循环（Linux/macOS）
   • 使用 httptools HTTP 解析器
   • 调整系统网络参数
   • 监控应用性能指标

4. 安全配置

   • 使用强密码和证书
   • 配置安全头
   • 限制访问权限
   • 定期更新依赖

14.3 常见使用场景 #

1. FastAPI 应用部署

   uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

2. Django 3+ 异步应用

   uvicorn myproject.asgi:application --host 0.0.0.0 --port 8000

3. Starlette 应用

   uvicorn myapp:app --host 0.0.0.0 --port 8000

4. 自定义 ASGI 应用

   import uvicorn
   
   async def app(scope, receive, send):
       # 你的 ASGI 应用逻辑
       pass
   
   if __name__ == "__main__":
       uvicorn.run(app, host="0.0.0.0", port=8000)

14.4 学习资源 #

1. 官方文档

   • Uvicorn 官方文档
   • ASGI 规范

2. 相关框架

   • FastAPI 文档
   • Django 文档
   • Starlette 文档

3. 部署和运维

   • Gunicorn 文档
   • Nginx 文档
   • Docker 文档