如何在Jupyter Notebook中用AsyncSSH+IPyWidget实时并行显示远程脚本输出

问题背景 #

需要在Jupyter Notebook中用多个IPyWidget Output组件，实时并行监控多台远程机器上Python脚本的输出，但现有基于AsyncSSH的方案无法实现实时更新。尝试过线程、纯asyncio方案，认为while循环是关键，但不知道如何正确集成到现有代码中。

问题根源 #

1. 远程脚本的print输出在非交互式环境下会被系统缓冲，导致输出不会实时发送到SSH通道，而是等到脚本结束或缓冲区满才一次性输出。
2. Jupyter的asyncio事件循环需要明确让出执行权，才能确保UI组件实时更新，否则会被长时间阻塞。

解决方案 #

步骤1：修改远程脚本，强制实时输出 #

修改print_hostname.py，给所有print添加flush=True参数，禁用输出缓冲：

import time

print("Début du script sur la machine.", flush=True)

for i in range(5):
    print(f"Étape {i} sur la machine.", flush=True)
    time.sleep(4)

print("Fin du script sur la machine.", flush=True)

步骤2：调整本地AsyncSSH代码，集成异步循环 #

修改本地代码，在SSH会话等待过程中加入异步循环，主动让出事件循环确保UI实时更新，同时优化Output组件的输出方式：

import asyncssh
import asyncio
import sys
from ipywidgets import Output, HBox

class MySSHClientSession(asyncssh.SSHClientSession):
    def __init__(self, output_widget):
        super().__init__()
        self._output_widget = output_widget
        self._closed = False

    def data_received(self, data, datatype):
        # 使用上下文管理器确保UI正确刷新
        with self._output_widget:
            if datatype == asyncssh.EXTENDED_DATA_STDERR:
                print(data.decode(), end='', file=sys.stderr)
            else:
                print(data.decode(), end='')

    def connection_lost(self, exc):
        self._closed = True
        if exc:
            with self._output_widget:
                print(f"SSH session error: {exc}", file=sys.stderr)

async def run_remote_command(host, username, password, command, output_widget):
    try:
        async with asyncssh.connect(host, username=username, password=password, known_hosts=None) as conn:
            chan, session = await conn.create_session(lambda: MySSHClientSession(output_widget), command)
            
            # 异步循环等待会话关闭，每次循环让出事件循环给UI更新
            while not session._closed:
                await asyncio.sleep(0.1)
            
            await chan.wait_closed()
    except Exception as e:
        with output_widget:
            print(f"Error connecting to {host}: {str(e)}", file=sys.stderr)

async def main():
    # 假设parse_creds是你已实现的主机信息解析函数
    host_infos = [parse_creds(i) for i in range(6)]
    cmds = [f"python /scripts/print_hostname.py {P}" for P in range(1,7)]
    outputs = [Output(layout={'border': '1px solid white', 'width': '200px'}) for _ in host_infos]
    
    tasks = [run_remote_command(host_info[0], host_info[1], host_info[2], cmd, out) 
             for host_info, cmd, out in zip(host_infos, cmds, outputs)]
    
    display(HBox(outputs))
    await asyncio.gather(*tasks)

# 在Jupyter中直接运行，替代asyncio.create_task更稳定
await main()

关键改进说明 #

• 缓冲处理：远程脚本通过flush=True强制实时输出，避免系统缓冲延迟。
• 异步循环：while not session._closed循环配合asyncio.sleep(0.1)，主动让出事件循环，让Jupyter有机会更新UI组件。
• UI刷新优化：改用with output_widget:上下文管理器输出内容，比append_stdout/append_stderr更可靠，确保输出实时渲染。
• Jupyter运行方式：直接用await main()替代asyncio.create_task，在Jupyter的异步环境中调度更稳定。

内容的提问来源于stack exchange，提问作者Myoch