如何使用telnetlib3与Python读取Telnet输出并设置超时避免阻塞？

你的代码阻塞是因为reader.read()会一直等待服务器发送数据，只要连接没断开，它就不会返回空字符串，导致while循环无限执行。要添加超时机制，可通过asyncio.wait_for()为读取操作或整个会话设置超时时间，到时间后主动停止读取并返回已收集的内容。

修改后的代码（适配Python 3.8） #

import telnetlib3
import asyncio

async def _read_telnet_output(reader, total_output):
    while True:
        output = await reader.read(4096)
        if output:
            print(output)
            total_output += output
        else:
            break

async def run_telnet_session():
    total_output = []
    # 建立Telnet连接，设置连接超时
    reader, writer = await telnetlib3.open_connection("192.168.200.10", 9000, timeout=5)
    
    try:
        # 给整个读取流程设置10秒超时
        read_task = asyncio.create_task(_read_telnet_output(reader, total_output))
        await asyncio.wait_for(read_task, timeout=10)
    except asyncio.TimeoutError:
        print("读取超时，终止会话")
    finally:
        writer.close()
        await writer.wait_closed()
    return ''.join(total_output)

output = asyncio.run(run_telnet_session())
print("最终收集的输出：", output)

关键修改说明 #

• 连接超时：在open_connection中添加timeout参数，避免连接阶段长时间阻塞。
• 会话总超时：用asyncio.wait_for()包装读取任务，设置整个Telnet会话的最大持续时间，超时后抛出TimeoutError，捕获异常后终止读取流程。
• 输出收集优化：改用列表收集输出片段，比字符串拼接更高效。

如果需要更精细的控制，也可以给单次reader.read()操作单独设置超时，替换读取函数里的逻辑即可：

async def _read_telnet_output(reader, total_output):
    while True:
        try:
            # 单次读取等待2秒超时
            output = await asyncio.wait_for(reader.read(4096), timeout=2)
            if output:
                print(output)
                total_output += output
            else:
                break
        except asyncio.TimeoutError:
            print("单次读取超时，终止会话")
            break

这样修改后，不管服务器是否保持连接，超时时间到后函数都会停止读取并返回已收集的内容。

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