第四十二篇：ThreadLocal 与分布式进程——并发进阶
前言
在并发编程中，线程本地存储（ThreadLocal）和分布式进程处理是两大实用技术。ThreadLocal 让每个线程拥有独立的数据副本，避免共享状态带来的同步问题；分布式进程则通过 managers 将计算任务分发到多台机器上运行。本文将深入讲解这两大高级并发技术。
ThreadLocal —— 线程局部变量
基本原理
import threading
import time
# 普通全局变量 — 所有线程共享（需要加锁保护）
shared_counter = 0
def unsafe_increment():
global shared_counter
for _ in range(1000):
shared_counter += 1 # 非原子操作，有竞态条件!
threads = [threading.Thread(target=unsafe_increment) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(f"全局变量: {shared_counter}") # 通常不是 10000!
# ThreadLocal — 每个线程独立的变量副本
thread_local = threading.local()
def safe_with_threadlocal(index):
thread_local.counter = 0
for _ in range(1000):
thread_local.counter += 1
print(f"线程-{index}: counter={thread_local.counter}")
threads = [threading.Thread(target=safe_with_threadlocal, args=(i,)) for i in range(5)]
for t in threads: t.start()
for t in threads: t.join()
# 每个线程的 counter 都是 1000，互不干扰!
实战：数据库连接池模拟
import threading
import time
class ConnectionPool:
"""线程安全的数据库连接池 — 使用 ThreadLocal"""
def __init__(self, pool_size=5):
self.pool_size = pool_size
self._local = threading.local() # 每个线程一个连接
def get_connection(self):
"""获取当前线程的数据库连接"""
if not hasattr(self._local, 'connection'):
self._local.connection = self._create_connection()
return self._local.connection
def _create_connection(self):
"""创建新连接（模拟耗时操作）"""
thread_id = threading.current_thread().ident
connection_id = f"conn_{id(self)}_{thread_id}"
print(f"[{threading.current_thread().name}] 创建新连接: {connection_id}")
return {"id": connection_id, "status": "open", "ping_count": 0}
def close_connection(self):
"""关闭当前线程的连接"""
if hasattr(self._local, 'connection'):
self._local.connection["status"] = "closed"
del self._local.connection
def ping(self):
"""发送心跳检测"""
conn = self.get_connection()
conn["ping_count"] += 1
return True
pool = ConnectionPool(pool_size=5)
def db_operation(thread_name, duration):
"""模拟数据库操作"""
conn = pool.get_connection()
print(f"[{thread_name}] 连接 ID: {conn['id']}")
pool.ping()
time.sleep(duration)
pool.close_connection()
print(f"[{thread_name}] 完成!")
threads = [
threading.Thread(target=db_operation, args=("Worker-A", 1)),
threading.Thread(target=db_operation, args=("Worker-B", 0.5)),
threading.Thread(target=db_operation, args=("Worker-C", 0.8)),
]
for t in threads: t.start()
for t in threads: t.join()
print("\n所有线程完成，各自持有独立的数据库连接!")
ThreadLocal 字典模式
import threading
import contextlib
@contextlib.contextmanager
def thread_local_context(**kwargs):
"""上下文管理器：临时设置 ThreadLocal 值"""
local_data = threading.local()
old_values = {}
for key in kwargs:
old_values[key] = getattr(local_data, key, None)
setattr(local_data, key, kwargs[key])
try:
yield local_data
finally:
for key, old_value in old_values.items():
if old_value is None:
delattr(local_data, key)
else:
setattr(local_data, key, old_value)
# 使用示例
def trace_request(request_id, user_id):
with thread_local_context(request_id=request_id, user_id=user_id) as ctx:
print(f"处理请求 {ctx.request_id}, 用户: {ctx.user_id}")
simulate_work()
def simulate_work():
ctx = threading.local()
request_id = getattr(ctx, "request_id", "unknown")
user_id = getattr(ctx, "user_id", "anonymous")
print(f" → 子方法中也能访问: {request_id}/{user_id}")
trace_request("req-123", "user-alice")
# 处理请求 req-123, 用户: user-alice
# → 子方法中也能访问: req-123/user-alice
logging 中的 ThreadLocal
import threading
import logging
class ThreadTraceFilter(logging.Filter):
def filter(self, record):
ctx = threading.local()
record.trace_id = getattr(ctx, "trace_id", "no-trace")
record.user_id = getattr(ctx, "user_id", "anon")
return True
logger = logging.getLogger("app")
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(
"%(asctime)s [%(trace_id)s][%(user_id)s] %(levelname)-7s %(message)s",
))
handler.addFilter(ThreadTraceFilter())
logger.addHandler(handler)
def process_order(order_id, customer):
ctx = threading.local()
ctx.trace_id = f"TRACE-{order_id}"
ctx.user_id = customer
logger.info(f"开始处理订单 {order_id}")
threads = [
threading.Thread(target=process_order, args=(1001, "Alice")),
threading.Thread(target=process_order, args=(1002, "Bob")),
]
for t in threads: t.start()
for t in threads: t.join()
multiprocessing.managers —— 分布式进程
基础用法
import multiprocessing
from multiprocessing.managers import BaseManager
import time
class TaskManager(BaseManager):
pass
TaskManager.register('submit_task')
TaskManager.register('get_result')
# 服务端
def server(host='localhost', port=0):
manager = TaskManager(address=('', port))
server = manager.get_server()
print(f"服务器启动于端口 {port}")
server.serve_forever()
# 客户端
def client(host='localhost', port=0):
manager = TaskManager(address=(host, port))
manager.connect()
manager.submit_task({"task": "计算平方", "value": 7})
result = manager.get_result()
print(result)
ProcessPoolExecutor 替代方案
import multiprocessing
from concurrent.futures import ProcessPoolExecutor, as_completed
import time
def cpu_intensive_task(x):
"""CPU 密集型计算"""
total = sum(i * i for i in range(x * 1_000_000))
return {"input": x, "output": total % (10**9 + 7)}
if __name__ == "__main__":
inputs = list(range(1, 7)) # 6 个任务
with ProcessPoolExecutor(max_workers=4) as executor:
futures = {executor.submit(cpu_intensive_task, inp): inp for inp in inputs}
for future in as_completed(futures):
input_val = futures[future]
try:
result = future.result()
print(f"输入 {input_val}: 输出 {result['output']}")
except Exception as e:
print(f"任务 {input_val} 出错: {e}")
# map — 简洁的批量处理
start = time.perf_counter()
results = list(executor.map(cpu_intensive_task, range(1, 5)))
elapsed = time.perf_counter() - start
print(f"map 方式耗时: {elapsed:.2f}s")
实际场景：大规模数据处理
import multiprocessing
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import time
import random
from collections import defaultdict
def preprocess_batch(batch_id, documents):
"""预处理一批文档（分词、去停用词等）"""
processed = []
for doc in documents:
words = doc.lower().split()
cleaned = [w.strip('.,;!?()[]{}"\\'-') for w in words if len(w) > 2]
processed.append({"id": batch_id, "text": cleaned})
time.sleep(0.1) # 模拟处理时间
return processed
def build_index(processed_docs, shard_id):
"""构建倒排索引分片"""
index = defaultdict(list)
for doc in processed_docs:
for word in doc["text"]:
index[word].append(doc["id"])
return {"shard": shard_id, "terms": dict(list(index.items())[:50])}
random.seed(42)
vocab = ["python", "java", "javascript", "rust", "go", "typescript"]
documents = []
for i in range(100):
num_words = random.randint(5, 20)
words = [random.choice(vocab) for _ in range(num_words)]
documents.append(" ".join(words))
# 第一阶段：并行预处理（I/O 密集 → 多线程）
start = time.perf_counter()
batch_size = 10
with ThreadPoolExecutor(max_workers=4) as executor:
preprocessed_batches = list(executor.map(
preprocess_batch,
range(len(documents)//batch_size + 1),
[documents[i:i+batch_size] for i in range(0, len(documents), batch_size)]
))
preprocess_time = time.perf_counter() - start
print(f"预处理耗时: {preprocess_time:.2f}s")
# 第二阶段：并行构建索引（CPU 密集 → 多进程）
flat_preprocessed = [doc for batch in preprocessed_batches for doc in batch]
start = time.perf_counter()
with ProcessPoolExecutor(max_workers=4) as executor:
shards = list(executor.map(build_index, flat_preprocessed, range(4)))
shard_time = time.perf_counter() - start
print(f"索引构建耗时: {shard_time:.2f}s")
两种方案对比与选型
| 方案 | 优势 | 局限 | 适用场景 |
|---|---|---|---|
| ThreadLocal | 线程隔离，无需锁 | 仅限同一进程内的线程 | Web 请求上下文、DB 连接池 |
| multiprocessing.Pool | 简单易用，自动负载均衡 | 仅本地多核 | 批量 CPU 计算 |
| multiprocessing.managers | 支持网络共享对象 | 序列化开销大 | 共享配置表、小型分布式任务 |
| concurrent.futures | 统一接口，自动切换 | 功能相对精简 | 通用并发需求 |
决策树
需要利用多核 CPU 吗？
├─ 否 → 多线程（ThreadPoolExecutor）
└─ 是
需要跨机器扩展吗？
├─ 否 → 多进程（ProcessPoolExecutor）
└─ 是 → multiprocessing.managers / 外部存储 + 分布式框架
最佳实践总结
✅ 推荐做法
- 优先使用 concurrent.futures — 统一的 Executor 接口，代码易迁移
- ThreadLocal 存储请求上下文 — Web 框架中保存 request/session/user 信息
- 避免过度复杂的分布式架构 — 先用 Pool，再考虑 managers
- 注意 pickle 安全性 — 只信任可信来源的序列数据
- 合理设置 max_workers — CPU 密集型 = 核心数，I/O 密集型 ≈ 核心数 × 10~20
❌ 避免陷阱
import threading
import multiprocessing
# ❌ 在线程间传递 ThreadLocal 数据（不能跨线程/进程共享）
tl = threading.local()
tl.value = "hello"
def copy_value(tl_source):
other_tl = threading.local()
# other_tl.value = tl_source.value ← AttributeError!
# ✅ 如需共享数据，使用 Queue/Pipe/共享内存
# ❌ 在大进程中传递大型闭包（会序列化整个对象）
large_dict = {i: f"value_{i}" for i in range(1_000_000)}
p = multiprocessing.Process(target=lambda: large_dict[0])
# ✅ 传递必要参数，或在子进程中重新构建数据
p = multiprocessing.Process(target=function_with_arg, args=(large_dict[0],))
# ❌ 忽略 KeyboardInterrupt 导致僵尸进程
try:
with ProcessPoolExecutor() as executor:
executor.map(func, items)
except KeyboardInterrupt:
pass # 子进程可能仍在运行!
# ✅ 正确清理
try:
with ProcessPoolExecutor() as executor:
executor.map(func, items)
except KeyboardInterrupt:
pass # 退出 with 时会自动终止子进程
总结
本文系统讲解了 Python 并发进阶的高级特性：
- ThreadLocal — 线程隔离数据存储、连接池模拟、Logging 上下文追踪
- ThreadLocal 字典模式 — 上下文管理器封装，跨函数链式访问
- multiprocessing.managers — 共享数据结构、网络通信、多节点协调
- ProcessPoolExecutor — 统一并发 API、as_completed 流式处理、大数据管道
- 分布式架构选择 — 决策树指导方案设计
- 最佳实践 — 选择策略、安全注意事项、资源清理
本文是 Python 教程系列的第四十二篇。