在 Jupyter Notebook 中实现并行任务队列，以在不阻塞 Notebook 界面的情况下执行耗时较长的函数。通过使用 `concurrent.futures.ThreadPoolExecutor` 和 `ipywidgets.Output`，可以实现任务的异步执行和结果的实时显示，同时避免了常见的序列化和作用域问题。## 使用 ThreadPoolExecutor 实现并行任务在 Jupyter Notebook 中并行执行任务，同时保持 Notebook 的响应性，是一个常见的需求。`concurrent.futures.ThreadPoolExecutor` 提供了一种简单有效的方法来实现这一点。结合 `ipywidgets.Output`，我们可以将任务的输出实时显示在 Notebook 中。### 核心代码以下代码展示了如何使用 `ThreadPoolExecutor` 和 `ipywidgets.Output` 来实现并行任务队列：“`pythonimport sysimport asyncioimport concurrent.futuresimport ipywidgetsthreadpool = 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 函数是核心，它负责将任务提交到线程池并处理输出。
out = ipywidgets.Output(): 创建一个 Output 控件，用于显示任务的输出。print 函数重定向: 重新定义 print 函数，使其将输出发送到 Output 控件。done 函数: 定义一个回调函数，在任务完成后执行，用于处理任务的结果或异常。go 协程: 创建一个异步协程，使用 loop.run_in_executor 将任务提交到线程池。asyncio.create_task: 创建一个 asyncio 任务来运行 go 协程。task.add_done_callback: 将 done 函数添加到任务的回调列表中，以便在任务完成后执行。
示例用法
以下代码展示了如何使用 run 函数来执行一个耗时的函数：
import timedef 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 中实时显示任务的输出。
以上就是# 并行执行 Jupyter Notebook 中的任务队列的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1367760.html
微信扫一扫 
支付宝扫一扫 
