在 Jupyter Notebook 中实现并行评估队列

本文将探讨如何在 Jupyter Notebook 环境中实现并行任务处理，以避免长时间运行的函数阻塞 Notebook 的交互体验。我们将利用 Python 的 concurrent.futures 模块和 ipywidgets 库，创建一个可以在后台执行任务的线程池，并将任务的输出实时显示在 Notebook 中。

使用 concurrent.futures.ThreadPoolExecutor 实现并行

concurrent.futures 模块提供了一个高级接口，用于异步执行可调用对象。其中，ThreadPoolExecutor 类允许我们在线程池中执行任务，从而实现并行处理。

使用 ipywidgets.Output 显示任务输出

ipywidgets 库提供了一系列交互式控件，可以方便地在 Notebook 中显示和操作数据。Output 控件可以捕获标准输出和标准错误，并将其显示在 Notebook 中，这对于显示后台任务的执行进度和结果非常有用。

实现代码

以下代码展示了如何使用 concurrent.futures.ThreadPoolExecutor 和 ipywidgets.Output 来实现并行任务处理：

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

这段代码定义了一个 run 函数，它接受一个可调用对象 fn 和一些参数，并在线程池中执行该对象。run 函数还创建了一个 ipywidgets.Output 控件，用于显示任务的输出。

在 run 函数中，我们定义了一个 print 函数，它将输出重定向到 Output 控件。我们还定义了一个 done 函数，它在任务完成后被调用，用于显示任务的完成状态。

最后，我们使用 asyncio.create_task 创建一个异步任务，并在线程池中执行该任务。

示例

以下代码展示了如何使用 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 函数，它模拟一个长时间运行的任务。run 函数将 cpu_bound 函数放入线程池执行，并将输出显示在 Notebook 中。

注意事项

• 线程池的大小应该根据实际情况进行调整。如果线程池过小，可能会导致任务排队等待执行；如果线程池过大，可能会导致系统资源耗尽。
• 在执行长时间运行的任务时，应该注意任务的资源消耗，避免占用过多的 CPU 和内存资源。
• ipywidgets.Output 控件依赖于 asyncio 循环，确保在 Jupyter Notebook 中正确运行。

总结

通过使用 concurrent.futures.ThreadPoolExecutor 和 ipywidgets.Output，我们可以在 Jupyter Notebook 中实现并行任务处理，而不会阻塞 Notebook 本身。这种方法可以提高 Notebook 的交互体验，并允许我们同时执行多个任务。