从Python 2.7 telnetlib迁移至Python 3 telnetlib3：asyncio带来诸多困惑

问题背景 #

原Python 2.7脚本使用telnetlib实现基础设备的Telnet交互，升级至Python 3.x时，telnetlib已被标记为弃用，官方推荐使用telnetlib3替代。但telnetlib3基于asyncio异步机制，与原同步多线程逻辑差异较大，缺少简单的命令发送/结果读取示例，导致迁移困难。

原Python 2.7代码 #

import telnetlib

# Variables
telnetConnection = None
password = "password"

# Open connection
try:
    telnetConnection = telnetlib.Telnet('192.168.1.123', 23, 5)
except Exception:
    print "UNABLE TO CONENCT"

# Send/Recieve Commands
if telnetConnection != None:
    # Send 'ACCESS' command
    telnetConnection.write("ACCESS" + "\r" + "\n")

    # Wait for 'Passord: ?' prompt
    telnetConnection.read_until("Password: ?", 10)

    # Send password
    telnetConnection.write(password + "\r" + "\n")

    # Wait for '=>' prompt
    passReturn = telnetConnection.read_until("=>", 10)

    # Check to make sure password was accepted
    if 'Invalid Password' in passReturn:
        print 'Invalid password'
    else:
        # Send 'CLEAR' command
        telnetConnection.write("CLEAR" + "\r" + "\n")

        # Wait for '?' prompt
        telnetConnection.read_until("?", 10)

        # Send 'Y' command
        telnetConnection.write("Y" + "\r" + "\n")

        # Wait for '=>' prompt
        telnetConnection.read_until("=>", 10)

        # Close connection
        telnetConnection.close()

对应的Python 3 telnetlib3实现 #

单设备异步交互代码 #

import asyncio
import telnetlib3

async def telnet_device(host, port, password):
    try:
        # 建立Telnet连接，对应原telnetlib.Telnet初始化
        reader, writer = await telnetlib3.open_connection(host, port, timeout=5)
    except Exception:
        print("UNABLE TO CONNECT")
        return

    try:
        # 发送ACCESS命令，writer.write替代原write，drain确保数据发送完成
        writer.write("ACCESS\r\n")
        await writer.drain()

        # 等待Password: ?提示，readuntil需await等待异步结果
        await reader.readuntil("Password: ?", timeout=10)

        # 发送密码
        writer.write(f"{password}\r\n")
        await writer.drain()

        # 等待=>提示并获取返回内容
        pass_return = await reader.readuntil("=>", timeout=10)

        # 校验密码是否有效
        if 'Invalid Password' in pass_return:
            print('Invalid password')
            return

        # 发送CLEAR命令
        writer.write("CLEAR\r\n")
        await writer.drain()

        # 等待?提示
        await reader.readuntil("?", timeout=10)

        # 发送Y确认命令
        writer.write("Y\r\n")
        await writer.drain()

        # 等待操作完成后的=>提示
        await reader.readuntil("=>", timeout=10)

    finally:
        # 关闭连接，需等待连接关闭完成
        writer.close()
        await writer.wait_closed()

# 执行单设备交互
if __name__ == "__main__":
    asyncio.run(telnet_device('192.168.1.123', 23, "password"))

多设备并发处理（替代原多线程） #

原多线程逻辑可通过asyncio.gather实现异步并发，无需彻底重写核心交互逻辑：

async def main():
    # 定义多设备列表，格式为(主机地址, 端口, 密码)
    devices = [
        ('192.168.1.123', 23, "password"),
        ('192.168.1.124', 23, "password"),
        # 可添加更多设备
    ]
    # 并发执行所有设备的Telnet任务
    await asyncio.gather(*[telnet_device(host, port, pwd) for host, port, pwd in devices])

if __name__ == "__main__":
    asyncio.run(main())

关键差异说明 #

• 异步机制：所有IO操作（连接、读取、发送后确认）需加await关键字，函数需定义为async def
• 读写分离：telnetlib3通过reader（读取）和writer（写入）两个对象替代原Telnet实例
• 发送确认：writer.write后需调用await writer.drain()确保数据发送至设备
• 多设备处理：用asyncio.gather替代多线程，实现更高效的异步并发，核心交互逻辑可复用

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