第 10 章 部署与并发

目录介绍

• 10.1 打包与分发
  • 10.1.1 setuptools 打包
  • 10.1.2 发布到 PyPI
  • 10.1.3 pyinstaller 打包
  • 10.1.4 Docker 容器化
  • 10.1.5 版本管理策略
• 10.2 并发编程
  • 10.2.1 GIL 简介
  • 10.2.2 多线程用法
  • 10.2.3 多进程用法
  • 10.2.4 线程与进程池
  • 10.2.5 asyncio 异步
  • 10.2.6 并发模型对比
• 10.3 新手陷阱
• 10.4 综合思考题

10.1 打包与分发

3.1.1 setuptools 打包 Python 包

把你的代码打包成 pip install 能安装的标准 Python 包：

my_utils/                          # 项目根目录
├── pyproject.toml                 # 打包配置（现代方案）
├── setup.py                       # 打包配置（传统方案——仍广泛使用）
├── README.md
├── LICENSE
├── src/
│   └── my_utils/                  # 包源码
│       ├── __init__.py
│       ├── core.py
│       └── helpers.py
└── tests/
    └── test_core.py

📁 pyproject.toml（现代——推荐）：

[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "my-utils"
version = "0.1.0"
description = "一个实用的 Python 工具库"
readme = "README.md"
license = {text = "MIT"}
authors = [{name = "杨充", email = "yangchong@example.com"}]
requires-python = ">=3.8"
classifiers = [
    "Programming Language :: Python :: 3",
    "License :: OSI Approved :: MIT License",
    "Operating System :: OS Independent",
]
dependencies = [
    "requests>=2.28",
    "pandas>=1.5",
]

[project.optional-dependencies]
dev = ["pytest>=7.0", "black>=23.0", "mypy>=1.0"]

[project.urls]
Homepage = "https://github.com/yangchong211/my-utils"

📁 setup.py（传统——兼容性好）：

from setuptools import setup, find_packages

setup(
    name="my-utils",
    version="0.1.0",
    packages=find_packages(where="src"),
    package_dir={"": "src"},
    install_requires=["requests>=2.28"],
    python_requires=">=3.8",
)

# ===== 打包命令 =====
pip install build twine           # 安装打包工具

# 构建——生成 dist/ 目录
python -m build
# dist/
# ├── my_utils-0.1.0-py3-none-any.whl   ← wheel 格式（现代，推荐）
# └── my_utils-0.1.0.tar.gz              ← 源码格式

# 本地安装测试
pip install dist/my_utils-0.1.0-py3-none-any.whl
# 或可编辑安装（开发中——改代码即时生效）
pip install -e .

3.1.2 发布到 PyPI

把你的包分享给全世界：

# 1. 注册 PyPI 账号：https://pypi.org
# 2. 创建 API Token：https://pypi.org/manage/account/token/

# 3. 配置 ~/.pypirc（可选——或用环境变量）
# [pypi]
# username = __token__
# password = pypi-xxxxxxxxxxxx

# 4. 发布到 PyPI
twine upload dist/*

# 发布到 TestPyPI（先测试）
twine upload --repository testpypi dist/*
pip install --index-url https://test.pypi.org/simple/ my-utils

3.1.3 pyinstaller 打包 exe

把 Python 脚本打包成独立可执行文件——给没有 Python 环境的用户使用：

pip install pyinstaller

# 基本用法——生成单个文件夹
pyinstaller main.py
# → dist/main/   （文件夹模式——启动最快）

# 打包成单个 exe 文件
pyinstaller --onefile main.py
# → dist/main.exe （单文件模式——方便分发，启动稍慢）

# 隐藏控制台窗口（GUI 程序）
pyinstaller --onefile --noconsole main.py

# 指定图标
pyinstaller --onefile --icon=app.ico main.py

# 添加数据文件（配置文件、图片等）
pyinstaller --onefile --add-data "config.json;." main.py

⚠️ pyinstaller 常见问题：

# 1. 文件太大？排除不必要的模块
pyinstaller --onefile --exclude-module matplotlib --exclude-module numpy main.py

# 2. 找不到动态库？
# macOS: pyinstaller 可能签名问题 → codesign --force --deep --sign - dist/main
# Windows: 检查 .dll 是否在 PATH 里

# 3. 跨平台限制？
# macOS 上打包的 exe ← 不能在 Windows 上运行
# 需要在目标平台上分别打包

3.1.4 Docker 容器化部署

Docker 解决"在我机器上能跑"的问题——把 Python 环境 + 代码一起打包成镜像：

📁 Dockerfile：

# ===== 基础镜像（越小越好——推荐 slim） =====
FROM python:3.12-slim

# ===== 设置工作目录 =====
WORKDIR /app

# ===== 安装系统依赖（某些 Python 包需要 C 编译器）=====
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# ===== 先复制依赖文件（利用 Docker 缓存——这一步不变就不重建） =====
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# ===== 再复制代码（经常变——放最后）=====
COPY src/ src/

# ===== 非 root 用户运行（安全） =====
RUN useradd --create-home appuser
USER appuser

# ===== 暴露端口 =====
EXPOSE 8000

# ===== 启动命令 =====
CMD ["python", "-m", "uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

📁 .dockerignore：

__pycache__
*.pyc
.venv
.git
.env
*.md

# 构建镜像
docker build -t my-app:v1.0 .

# 运行容器
docker run -d -p 8000:8000 --name my-app my-app:v1.0

# 查看日志
docker logs -f my-app

# 多阶段构建——构建环境 vs 运行环境（更小）
# 第一阶段：编译
FROM python:3.12 AS builder
COPY requirements.txt .
RUN pip install --user -r requirements.txt

# 第二阶段：运行（不含编译器——更小更安全）
FROM python:3.12-slim
COPY --from=builder /root/.local /home/appuser/.local
COPY src/ src/
ENV PATH=/home/appuser/.local/bin:$PATH
# ...

🔑 Docker Compose 多服务编排：

# docker-compose.yml
version: "3.9"
services:
  web:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
    depends_on:
      - db
    restart: unless-stopped

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

3.1.5 版本管理策略

# ===== 语义化版本（Semantic Versioning） =====
# MAJOR.MINOR.PATCH
# 1  .  2  .  3
# │     │     └─ PATCH：向后兼容的 bug 修复
# │     └─────── MINOR：向后兼容的新功能
# └───────────── MAJOR：不兼容的 API 变更

# 示例：
# 1.0.0 → 初始版本
# 1.0.1 → 修复了一个 bug
# 1.1.0 → 新增了一个功能，所有现有 API 仍然可用
# 2.0.0 → 删除了某个 API，破坏了向后兼容

# ===== 版本号管理工具 =====
pip install bump2version

# pyproject.toml 里初始版本：version = "0.1.0"
bump2version patch   # → 0.1.1
bump2version minor   # → 0.2.0
bump2version major   # → 1.0.0

# ===== Git Tag 管理 =====
git tag v1.0.0
git push --tags

10.2 并发编程

3.2.1 GIL 是什么？为什么 Python 有它？

🔑 GIL（Global Interpreter Lock） = 一把全局锁——同一时刻只有一个线程在执行 Python 字节码。

线程 1 ─→ [等待 GIL] ─→ [获得 GIL: 执行] ─→ [释放 GIL] ─→ [等待] ─→ ...
                                          ↓
线程 2 ─→ [等待 GIL] ─→ [等待...] ─→ [获得 GIL: 执行] ─→ [释放 GIL] ...

这导致一个反直觉的现象——Python 多线程在 CPU 密集型任务上比单线程还慢：

import time
import threading

def cpu_bound_task():
    """纯计算任务——CPU 密集型"""
    total = 0
    for i in range(50_000_000):
        total += i
    return total

# 单线程
start = time.perf_counter()
cpu_bound_task()
cpu_bound_task()
print(f"单线程：{time.perf_counter() - start:.3f}s")

# 多线程——期望快一倍，实际差不多甚至更慢！
start = time.perf_counter()
t1 = threading.Thread(target=cpu_bound_task)
t2 = threading.Thread(target=cpu_bound_task)
t1.start(); t2.start()
t1.join(); t2.join()
print(f"多线程：{time.perf_counter() - start:.3f}s")
# 单线程：5.2s
# 多线程：5.8s  ← 更慢！因为 GIL 竞争 + 上下文切换开销

🔑 为什么要 GIL？ ——Python 的内存管理（引用计数）不是线程安全的。没有 GIL，a = b 这样的赋值需要加锁——99% 的代码都要加锁，性能更差。

3.2.2 threading：多线程与 GIL 限制

多线程虽然受 GIL 限制——但 IO 密集型任务不受影响（因为 IO 等待时 GIL 自动释放）：

import threading
import time
import requests

def fetch_url(url: str, results: list, index: int):
    resp = requests.get(url, timeout=10)
    results[index] = len(resp.text)

urls = ["https://httpbin.org/delay/1"] * 5    # 5 个 1 秒的请求

# 单线程——串行：5 秒
start = time.perf_counter()
results_single = [0] * len(urls)
for i, url in enumerate(urls):
    fetch_url(url, results_single, i)
print(f"单线程：{time.perf_counter() - start:.3f}s")   # ~5s

# 多线程——并行：1 秒（IO 等待时 GIL 释放）
start = time.perf_counter()
results = [0] * len(urls)
threads = []
for i, url in enumerate(urls):
    t = threading.Thread(target=fetch_url, args=(url, results, i))
    threads.append(t)
    t.start()

for t in threads:
    t.join()
print(f"多线程：{time.perf_counter() - start:.3f}s")   # ~1s——快了 5 倍！

🔑 线程安全——Lock：

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(1_000_000):
        with lock:            # 或 lock.acquire(); ...; lock.release()
            counter += 1

threads = [threading.Thread(target=increment) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)               # 4,000,000 ✅（不加锁结果随机错误）

3.2.3 multiprocessing：绕过 GIL 的多进程

多进程每个进程有自己的 Python 解释器 + 自己的 GIL——真正并行：

import multiprocessing
import time

def cpu_task(name, n):
    """CPU 密集型——多进程才能真正加速"""
    total = 0
    for i in range(n):
        total += i ** 2
    print(f"[{name}] 完成，total={total}")
    return total

# 单进程
start = time.perf_counter()
cpu_task("main", 20_000_000)
cpu_task("main", 20_000_000)
print(f"单进程：{time.perf_counter() - start:.3f}s")

# 多进程
start = time.perf_counter()
p1 = multiprocessing.Process(target=cpu_task, args=("p1", 20_000_000))
p2 = multiprocessing.Process(target=cpu_task, args=("p2", 20_000_000))
p1.start(); p2.start()
p1.join(); p2.join()
print(f"多进程：{time.perf_counter() - start:.3f}s")
# 多进程几乎快一倍——因为真正并行

多进程通信：

# Queue——进程间安全通信
from multiprocessing import Process, Queue

def worker(q: Queue, index: int):
    result = sum(i ** 2 for i in range(1_000_000))
    q.put((index, result))

if __name__ == "__main__":
    q = Queue()
    processes = [Process(target=worker, args=(q, i)) for i in range(4)]
    for p in processes: p.start()
    for p in processes: p.join()

    results = {}
    while not q.empty():
        idx, val = q.get()
        results[idx] = val
    print(results)

# Pool——批量任务 + 自动负载均衡
from multiprocessing import Pool

def square(x):
    return x ** 2

with Pool(processes=4) as pool:
    results = pool.map(square, range(20))
print(results)   # 并行计算 20 个平方

3.2.4 concurrent.futures：线程池与进程池

最推荐的并发 API——统一接口，一行切换线程/进程：

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed
import time

def io_task(n: int) -> int:
    """IO 密集型——用线程池"""
    time.sleep(0.5)
    return n * 2

def cpu_task(n: int) -> int:
    """CPU 密集型——用进程池"""
    total = 0
    for i in range(5_000_000):
        total += i ** 2
    return total + n

# ===== 线程池：IO 密集型 =====
with ThreadPoolExecutor(max_workers=8) as executor:
    futures = [executor.submit(io_task, i) for i in range(10)]
    for future in as_completed(futures):
        print(future.result(), end=" ")   # 完成后立即获取（不是提交顺序）
# 10 个任务并行——总耗时 ~0.5s

# ===== 进程池：CPU 密集型 =====
with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(cpu_task, range(8)))
    print(f"\n进程池结果：{results[:3]}...")

🔑 map vs submit：

# map：按顺序返回——简单粗暴
with ThreadPoolExecutor() as ex:
    results = ex.map(fetch_url, urls)    # 等待全部完成，返回迭代器

# submit + as_completed：谁先完成谁先处理
with ThreadPoolExecutor() as ex:
    futures = {ex.submit(fetch_url, url): url for url in urls}
    for f in as_completed(futures):
        print(f"{futures[f]} → {f.result()}")  # 快的结果先出来

3.2.5 asyncio：异步 IO

asyncio 用单线程 + 事件循环处理大量 IO——性能比多线程更好，代码不需要锁：

import asyncio
import aiohttp                        # pip install aiohttp
import time

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.text()

async def main():
    urls = ["https://httpbin.org/delay/1"] * 5

    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        print(f"获取了 {len(results)} 个响应")

start = time.perf_counter()
asyncio.run(main())
print(f"耗时：{time.perf_counter() - start:.3f}s")   # ~1s——5 个 1 秒请求并行

async/await 核心语法：

关键字	作用
async def	定义协程函数——可以被 await
await	等待异步操作完成——暂停当前协程，让出控制权给事件循环
asyncio.gather(*tasks)	并发执行多个协程——等待全部完成
asyncio.create_task()	创建后台任务——不等待
asyncio.run()	顶层入口——启动事件循环

# asyncio 不是多线程——所有代码在单线程中执行
# "并发"靠的是：当任务 A 在等待 IO 时，切换到任务 B——
# 而不是"真正的并行"

async def task_a():
    print("A1")
    await asyncio.sleep(1)    # ← A 在这等着——事件循环去执行 B
    print("A2")

async def task_b():
    print("B1")
    await asyncio.sleep(0.5)
    print("B2")

# 输出：A1 → B1 → B2 → A2（B 先完成）
asyncio.run(asyncio.gather(task_a(), task_b()))

3.2.6 四种并发模型对比与选择

模型	适用场景	并行度	内存开销	通信方式
threading	IO 密集型（网络请求、文件读写）	伪并行（GIL）	低（共享进程内存）	共享变量 + Lock
multiprocessing	CPU 密集型（计算、图像处理）	真并行	高（独立进程）	Queue / Pipe / Manager
concurrent.futures	通用——高层封装	选线程池或进程池	按需	Future 对象
asyncio	高并发 IO（Web 服务器、爬虫）	单线程协作式	极低（一个线程）	Queue / 共享状态（无锁！）

🔑 决策树：

你的任务是？
  │
  ├─ IO 密集型（网络/文件）？
  │    │
  │    ├─ 并发数 < 100 → threading 或 ThreadPoolExecutor
  │    └─ 并发数 > 100 → asyncio（更快更省内存）
  │
  └─ CPU 密集型（计算/数据处理）？
       └─ multiprocessing 或 ProcessPoolExecutor

10.3 新手陷阱

#	陷阱	说明
1	多线程操作共享变量不加锁	counter += 1 不是原子操作（读→加→写三步）——多线程同时写导致数据丢失
2	multiprocessing 在 __main__ 外创建	Windows/macOS 必须放在 if __name__ == "__main__": 里——否则递归创建子进程
3	asyncio 里用 requests	requests 是同步的——会阻塞整个事件循环。用 aiohttp 或 httpx.AsyncClient
4	Docker 镜像用 python:latest	latest 随时变——生产环境用具体版本 python:3.12-slim
5	pyinstaller --onefile 启动慢	单文件每次启动都要解压到临时目录——频繁调用的脚本用文件夹模式

陷阱 2 详解：

# ❌ Windows 上直接写——递归创建子进程，死机
import multiprocessing
pool = multiprocessing.Pool(4)
pool.map(func, data)

# ✅ 加 __main__ 保护
if __name__ == "__main__":
    with multiprocessing.Pool(4) as pool:
        pool.map(func, data)

10.4 综合思考题

1. GIL 的未来：Python 3.13 引入了"Free-threaded CPython"（可选禁用 GIL）——你认为 GIL 最终会被移除吗？如果移除 GIL，现有的多线程代码（没有加锁但依赖 GIL 的代码）会崩溃吗？

2. Docker vs 虚拟环境：Docker 比 venv 重得多（镜像几百 MB vs venv 几十 MB）——在什么场景下 Docker 值得这个代价？在什么场景下 venv + systemd 就够了？

3. asyncio 的"传染性"：await 只能在 async def 函数里使用——一旦一个函数变成 async，所有调用它的函数都必须是 async。这种"传染"是设计缺陷还是有意为之？如果打破这个规则（在普通函数里 await）会怎样？

4. 多进程的序列化开销：multiprocessing 通过 pickle 序列化传递数据——如果传一个 500MB 的 DataFrame，它会被序列化（变成 ~1GB pickle）然后传给每个子进程。这比你想象的慢得多——有什么办法避免拷贝？multiprocessing.shared_memory 能解决问题吗？

5. 打包的"假跨平台"：pyinstaller 在 macOS 上打包的 exe 不能在 Windows 上运行——但 .whl 文件标注了 py3-none-any（any platform）。真正的跨平台 Python 分发要怎么做？Docker 和 WASM（WebAssembly）各有什么限制？