Python中使用telnetlib3实现Telnet连接读写报错及正确用法

错误原因 #

你遇到的RuntimeError本质是重复调用asyncio.run()创建了多个独立的事件循环：

• 第一次asyncio.run(telnetlib3.open_connection(...))会创建一个事件循环，返回的reader和writer对象绑定在这个循环上
• 第二次asyncio.run(reader.read())又创建了新的事件循环，尝试让绑定在旧循环的任务在新循环中执行，直接触发跨循环绑定的错误

正确实现代码 #

所有异步操作必须放在同一个协程函数内，只通过一次asyncio.run()启动整个流程：

import asyncio
import telnetlib3

async def telnet_query(host, port):
    # 建立Telnet连接（同循环内执行）
    reader, writer = await telnetlib3.open_connection(host, port)
    
    # 发送命令并确保数据出缓冲区
    writer.write("$SYS,INFO")
    await writer.drain()
    
    # 读取回复
    reply = await reader.read()
    print("收到回复:", reply)
    
    # 关闭连接释放资源
    writer.close()
    await writer.wait_closed()

if __name__ == "__main__":
    # 替换为你的主机和端口
    asyncio.run(telnet_query("your_host", your_port))

关键注意事项 #

• 所有需要await的异步操作必须归属于同一个事件循环，禁止拆分到多次asyncio.run()调用中
• 调用writer.write()后务必执行await writer.drain()，确保数据被实际发送到服务端，避免缓冲区积压
• 连接使用完毕后要主动关闭并等待关闭完成，防止资源泄漏

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