如何在 Python 中使用 Jupyter Kernel 运行代码并获取输出

在这篇文章中，我们将介绍如何在 Python 中使用jupyter_client包来启动 Jupyter Kernel，并通过它执行代码并获取输出。使用 Jupyter Kernel 是一种非常强大的方式，可以让你在 Python 程序中动态执行代码并获取结果。

下面是实现这一过程的详细代码，并附带中文注释，帮助大家更好地理解每个步骤。

安装依赖

首先，确保你已经安装了jupyter_client库。如果没有，可以使用以下命令安装：

pipinstalljupyter_client

Python 代码实现

importtimefromjupyter_clientimportKernelManagerdefmain():try:# 创建一个新的 KernelManager 实例，指定要使用的内核类型为 'python3'km=KernelManager(kernel_name='python3')km.start_kernel()# 创建一个客户端对象来与内核进行交互kc=km.client()# 启动与内核的通信通道kc.start_channels()# 确保客户端已连接到内核kc.wait_for_ready()# 准备要执行的代码，这里是简单的打印 "Hello, World!"code='print("Hello, World!")'# 执行代码并获取消息IDmsg_id=kc.execute(code)print("代码执行请求已发送。")# 等待并处理执行结果whileTrue:try:# 获取从内核发送过来的消息，设定超时时间为1秒msg=kc.get_iopub_msg(timeout=1)content=msg["content"]# 如果消息类型是流输出，并且输出的名字是 "stdout"，即标准输出ifmsg["msg_type"]=="stream"andcontent["name"]=="stdout":print(content["text"])# 打印内核执行结果break# 执行完成后退出循环exceptKeyboardInterrupt:print("用户中断了程序。")break# 用户中断时退出循环exceptExceptionase:# 捕获其他异常并输出print(f"发生异常：{e}")break# 清理资源，关闭通道print("清理资源，关闭通道...")kc.stop_channels()# 关闭内核print("关闭内核...")km.shutdown_kernel()exceptExceptionase:# 捕获整个过程中的任何异常print(f"遇到致命错误：{e}")finally:# 在任何情况下都能保证进行清理操作print("程序执行完成，进行最终清理...")# 如果该脚本作为主程序执行，调用 main() 函数if__name__=='__main__':main()

代码解读

启动内核：
我们使用KernelManager来管理 Jupyter 内核，并通过km.start_kernel()启动一个 Python 3 内核。内核用于执行我们传递给它的 Python 代码。
创建客户端：
KernelManager启动内核后，我们需要一个客户端来与内核进行交互。通过km.client()创建一个客户端对象kc。
启动通信通道：
使用kc.start_channels()启动与内核的通道。这是客户端与内核之间通信的基础。
等待内核准备就绪：
在执行代码之前，我们确保内核已经准备好接收命令，通过kc.wait_for_ready()来阻塞直到内核就绪。
执行代码并获取输出：
我们将要执行的代码放在code变量中，并通过kc.execute(code)来执行。该方法返回一个msg_id，该 ID 用于跟踪这次执行的消息。
获取输出结果：
通过kc.get_iopub_msg(timeout=1)获取从内核发来的消息。如果消息类型是 “stream” 并且内容的name为 “stdout”，我们就可以获取标准输出的内容，这就是我们打印的结果。
清理资源：
最后，我们停止与内核的通信通道并关闭内核，以释放资源。