在 jupyter Notebook 中实现并行任务队列，以在不阻塞 Notebook 界面的情况下执行耗时较长的函数。通过使用 `concurrent.futures.ThreadPoolExecutor` 和 `ipywidgets.Output`，可以实现任务的异步执行和结果的实时显示，同时避免了常见的序列化和作用域问题。 ## 使用 ThreadPoolExecutor 实现并行任务 在 Jupyter Notebook 中并行执行任务，同时保持 Notebook 的响应性，是一个常见的需求。`concurrent.futures.ThreadPoolExecutor` 提供了一种简单有效的方法来实现这一点。结合 `ipywidgets.Output`，我们可以将任务的输出实时显示在 Notebook 中。 ### 核心代码 以下代码展示了如何使用 `ThreadPoolExecutor` 和 `ipywidgets.Output` 来实现并行任务队列： “`python import sys import asyncio import concurrent.futures import ipywidgets threadpool = concurrent.futures.ThreadPoolExecutor(4) def run(fn, *args, **kwds): “run fn in threadpool” out = ipywidgets.Output() def print(*args, file=sys.stdout): line = ‘ ‘.join(map(str, args)) + ‘n’ if file is sys.stderr: out.append_stderr(line) else: out.append_stdout(line) def done(fut: asyncio.Future): try: result = fut.result() except asyncio.CancelledError: print(“cancelled”, fut, file=sys.stderr) except Exception: print(“failed”, fut, file=sys.stderr) else: print(“completed”, fut) async def go(): loop = asyncio.get_running_loop() return await loop.run_in_executor( threadpool, Lambda: fn(print, *args, **kwds), ) task = asyncio.create_task(go()) task.add_done_callback(done) return out
代码解释
-
导入必要的库:
- sys: 用于访问系统相关的参数和函数。
- asyncio: 用于编写并发代码。
- concurrent.futures: 用于异步执行任务。
- ipywidgets: 用于在 Jupyter Notebook 中创建交互式控件。
-
创建 ThreadPoolExecutor:
threadpool = concurrent.futures.ThreadPoolExecutor(4)
这行代码创建了一个包含 4 个线程的线程池。可以根据实际需求调整线程数量。
-
run 函数:
run 函数是核心，它负责将任务提交到线程池并处理输出。
示例用法
以下代码展示了如何使用 run 函数来执行一个耗时的函数：
import time def cpu_bound(print, dt, fail=False): for i in range(10): time.sleep(dt) print(i, time.time()) if fail: 1 / 0 return "done" run(cpu_bound, 0.1)
示例代码解释
- cpu_bound 函数模拟一个耗时的 CPU 密集型任务。它接受一个 print 函数作为参数，用于将输出发送到 Output 控件。
- run(cpu_bound, 0.1) 将 cpu_bound 函数提交到线程池，并传递 0.1 作为 dt 参数，表示每次循环暂停 0.1 秒。
错误处理
done 函数中包含了错误处理逻辑，可以捕获任务执行过程中发生的异常，并将错误信息输出到 Output 控件。
run(cpu_bound, 0.5, fail=True)
在这个例子中，fail=True 会导致 cpu_bound 函数抛出一个 ZeroDivisionError 异常，done 函数会捕获这个异常并将错误信息输出到 Output 控件。
注意事项
- 避免共享可变状态: 在多线程环境中，共享可变状态可能会导致数据竞争和死锁。尽量避免在任务之间共享可变状态。如果必须共享，请使用适当的同步机制，例如锁。
- 选择合适的线程池大小: 线程池的大小应该根据实际情况进行调整。如果任务是 CPU 密集型的，线程池的大小应该接近 CPU 的核心数量。如果任务是 IO 密集型的，线程池的大小可以大于 CPU 的核心数量。
- 序列化问题: 由于 multiprocessing 模块使用 pickle 进行序列化，因此某些对象可能无法被序列化。concurrent.futures.ThreadPoolExecutor 不存在此问题。
总结
通过使用 concurrent.futures.ThreadPoolExecutor 和 ipywidgets.Output，可以在 Jupyter Notebook 中轻松实现并行任务队列，从而提高 Notebook 的响应性和执行效率。这种方法避免了使用 multiprocessing 模块时可能遇到的序列化问题，并且可以在 Notebook 中实时显示任务的输出。
