本文详解如何使用 multiprocessing.dummy.Pool（即线程池）实现多线程并发执行任务，并将结果实时、线程安全地累积到共享列表中，同时修正常见语法与逻辑错误。
本文详解如何使用 multiprocessing.dummy.pool（即线程池）实现多线程并发执行任务，并将结果实时、线程安全地累积到共享列表中，同时修正常见语法与逻辑错误。
在 Python 中，多线程（multithreading）与多进程（multiprocessing）本质不同：前者共享内存空间，后者拥有独立内存。若目标是让多个线程协作更新同一份 self.state 列表（如实时收集 JSON 解析结果），则必须使用真正的多线程——而原代码中 from multiprocessing import Pool 实际启动的是多进程，不仅无法直接共享 self.state（子进程修改不会反映到主进程），还会因进程间通信开销降低 I/O 密集型任务的收益。
✅ 正确做法是切换至线程池：
from multiprocessing.dummy import Pool # 线程版 Pool（API 兼容） # 或等价写法： # from multiprocessing.pool import ThreadPool as Pool
该模块提供与 multiprocessing.Pool 完全一致的接口，但底层基于 threading.Thread，所有线程共享主线程的内存空间，因此 self.state.append(...) 可被多个线程直接调用——无需额外加锁（原因见下文注意事项）。
以下是修复后的完整可运行示例：
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
立即学习“Python免费学习笔记（深入）”；
from multiprocessing.dummy import Pool
def some_function(job_dict: dict) -> dict:
"""模拟网络请求+解析：此处仅返回输入，实际应包含 requests + json.loads 等 I/O 操作"""
import time
time.sleep(0.1) # 模拟网络延迟
return {"result": True, **job_dict}
class RunJobs:
def __init__(self, jobs: list[dict], threads=1):
self.jobs = jobs
self.threads = threads
self.state = [] # 所有线程共享此列表
def _run_sequentially(self):
for job in self.jobs:
result = some_function(job)
if result is not None: # 防止 None 被追加
self.state.append(result)
def _run_multithreaded(self):
with Pool(self.threads) as pool: # 推荐使用上下文管理器，自动 close/join
# 注意：args 必须是元组！(job,) 而非 (job) —— 后者等价于 job（括号无作用）
async_results = [pool.apply_async(some_function, args=(job,)) for job in self.jobs]
# 主线程等待全部完成并收集结果
for async_res in async_results:
result = async_res.get() # 阻塞直到完成
if result is not None:
self.state.append(result)
def run(self):
try:
if self.threads == 1:
self._run_sequentially()
else:
self._run_multithreaded()
except KeyboardInterrupt:
print("\nInterrupted. Current state:", self.state)
raise
# 使用示例
if __name__ == "__main__":
test_jobs = [
{"url": "https://httpbin.org/delay/1", "id": 1},
{"url": "https://httpbin.org/delay/1", "id": 2},
{"url": "https://httpbin.org/delay/1", "id": 3},
]
runner = RunJobs(test_jobs, threads=3)
runner.run()
print("Final collected results:", len(runner.state))? 关键要点说明：
- ✅ 线程安全追加（append）：CPython 的
list.append()是原子操作（GIL 保证），在多线程中直接调用self.state.append(x)是安全的，无需threading.Lock。 - ⚠️ 避免竞态条件的场景：若需
if x not in self.state: self.state.append(x)这类“检查-然后-操作”，则必须加锁——因为in和append是两个独立步骤。 - ? 不要混用 multiprocessing.Pool：它创建子进程，
self.state在各进程中为独立副本，主线程的self.state始终为空。 - ? I/O 密集型首选多线程：Web 抓取、文件读写等场景中，多线程能有效利用等待时间；CPU 密集型任务请用
multiprocessing。 - ? 资源管理最佳实践：使用
with Pool(...) as pool:自动处理清理，比手动close()+join()更健壮。
总结：要实现多线程共享状态更新，只需三步——换用 multiprocessing.dummy.Pool、确保 args 是合法元组（(job,)）、利用 GIL 保障 list.append 的线程安全性。这样即可高效、简洁地构建并发数据采集管道。
