pymongo 在多进程场景下频繁创建客户端会导致 windows 系统报 winerror 10048（端口耗尽），根本原因是每个子进程内反复新建 mongoclient 实例，引发大量短生命周期 tcp 连接堆积；正确做法是为每个工作进程复用单一客户端实例。
pymongo 在多进程场景下频繁创建客户端会导致 windows 系统报 winerror 10048（端口耗尽），根本原因是每个子进程内反复新建 mongoclient 实例，引发大量短生命周期 tcp 连接堆积；正确做法是为每个工作进程复用单一客户端实例。
在使用 multiprocessing 与 PyMongo 协同处理高并发数据库操作（如批量查询、更新）时，一个常见但隐蔽的陷阱是：看似“每个进程独立创建 MongoClient”，实则因调用频次过高导致底层 socket 资源快速耗尽。您遇到的 WinError 10048（“Only one usage of each socket address is normally permitted”）正是 Windows 系统对本地端口重用限制触发的典型错误——并非 MongoDB 服务异常，而是客户端连接管理失当。
问题核心在于：虽然 process_ticker_attention 中创建了 mongo_p，但其内部调用的 get_all_post_dates() 和 query_records_by_date() 各自又新建了独立的 MongoClient 实例（见原代码中两处 pymongo.MongoClient(...)）。假设单个 ticker 对应 2000 个日期，则每进程将创建至少 4000+ 次客户端连接（2 函数 × 2000 次循环），而 MongoClient 默认启用连接池且连接不会立即释放，极易突破 Windows 的临时端口范围（默认约 16384 个可用端口），尤其在 6 个进程并行时。
✅ 正确解决方案：利用 multiprocessing.Pool 的 initializer 机制，在每个子进程启动时初始化一次全局 MongoClient，并在整个进程生命周期内复用它。这既避免了重复建连开销，又彻底规避端口冲突。
以下是优化后的关键代码结构：
import multiprocessing
import pymongo
import sys
# 全局变量（仅在子进程中有效）
mongo_client = None
def init_worker():
"""子进程初始化函数：创建并缓存 MongoClient"""
global mongo_client
# 注意：connectTimeoutMS 和 socketTimeoutMS 应合理设置，避免阻塞
mongo_client = pymongo.MongoClient(
"mongodb://localhost:27017/",
connectTimeoutMS=30000,
socketTimeoutMS=30000,
maxPoolSize=10, # 控制每个进程最大连接数，防止资源过载
minPoolSize=1, # 保持最小连接，加速后续请求
serverSelectionTimeoutMS=5000
)
def get_all_post_dates(ticker):
"""复用全局 mongo_client，不再新建实例"""
global mongo_client
# 假设数据库名为 'StockForum'
db = mongo_client['StockForum']
# 示例聚合逻辑（请按实际需求替换）
pipeline = [
{"$match": {"ticker": ticker}},
{"$group": {"_id": "$date"}},
{"$project": {"date": "$_id", "_id": 0}}
]
return [doc["date"] for doc in db[ticker].aggregate(pipeline)]
def query_records_by_date(ticker, start_of_day):
"""同样复用全局 mongo_client"""
global mongo_client
db = mongo_client['StockForum']
return list(db[ticker].find({"date": start_of_day}))
def process_ticker_attention(ticker, finish_num, lock):
try:
ticker_dates = get_all_post_dates(ticker)
for date in ticker_dates:
data = query_records_by_date(ticker, date)
# 使用同一 client 更新目标集合
mongo_client['stock']['attention'].update_one(
{'date': date},
{'$set': {f'post_numbers.{ticker}': len(data)}}
)
with lock:
finish_num.value += 1
sys.stdout.write(f'\rfinish_count: {finish_num.value}')
sys.stdout.flush()
except Exception as e:
print(f'error in {ticker}: {e.__class__.__name__} - {e}')
if __name__ == '__main__':
# 主进程仅用于分发任务，不执行 DB 操作
database_name = 'StockForum'
# 使用 Manager 创建共享状态
with multiprocessing.Manager() as manager:
lock = manager.Lock()
finish_num = manager.Value('i', 0)
# 创建进程池，指定 initializer
with multiprocessing.Pool(
processes=6,
initializer=init_worker
) as pool:
# 获取 ticker 列表（主进程执行，避免跨进程传递大型对象）
mongo_main = pymongo.MongoClient("mongodb://localhost:27017/")
tickers = mongo_main[database_name].list_collection_names()
# 异步提交任务（注意：无需再传 mongo_client）
results = [
pool.apply_async(process_ticker_attention, (t, finish_num, lock))
for t in tickers
]
# 等待全部完成
for r in results:
r.get() # 可选：捕获子进程异常
? 关键注意事项：
- ❌ 禁止在任何被 apply_async 调用的函数内部新建 MongoClient —— 所有数据库操作必须通过 init_worker 初始化的全局 mongo_client 完成；
- ✅ maxPoolSize 是核心调优参数：设为 10 表示该进程最多维持 10 条活跃连接，足以应对并发读写，又不会过度占用端口；
- ⚠️ MongoClient 实例不可跨进程传递（会引发 Pickle 错误），必须在子进程内初始化；
- ? 如需更细粒度控制（如不同任务用不同数据库），可将 init_worker 改为接收参数（如 init_worker(db_uri, db_name)），配合 functools.partial 使用；
- ? Linux/macOS 用户虽较少遇 WinError 10048，但仍建议采用此模式——它显著提升性能并增强稳定性。
遵循这一模式后，您的多进程 PyMongo 应用将稳定运行于数千级文档处理场景，彻底告别端口耗尽错误。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
