如何用Python以非阻塞方式获取Telnet命令的完整响应？

在Windows系统使用Python 3.10.9，通过telnetlib创建Telnet会话时，执行命令后读取响应遇到以下问题：

• 使用session.read_some()等方法仅能获取部分响应，完整内容缺失超一半
• 尝试read_all()（阻塞）、read_eager()、read_very_eager()等方法均存在异常，例如read_very_eager()有时返回重复的分段内容

需求：

1. 非阻塞获取完整响应的实现方案
2. 更易用的Telnet替代模块

相关代码及示例：
创建会话：

session = telnetlib.Telnet(host, port, timeout)

发送命令：

session.write(command + b"\n")

完整响应示例：

Invalid arguments

Usage: $IMU,START,<SAMPLING_RATE>,<OUTPUT_RATE>
  where SAMPLING_RATE = [1 : 1000] in Hz
        OUTPUT_RATE   = [1 : SAMPLING_RATE] in Hz

实际读取到的内容：

b'\x1b[0GInvalid arguments\r\n\r\nUsage: $IMU,START,<'

一、非阻塞获取完整响应的方案 #

Telnet响应通常分批次传输，非阻塞场景下需要循环读取并判断响应是否结束，以下是两种可行方案：

1. 基于超时循环读取 #

通过read_eager()结合短超时循环，当连续一段时间读取不到新内容时，判定响应结束：

import time
import telnetlib

def get_full_telnet_response(session, check_interval=0.1, max_idle_time=2):
    response = b""
    last_update_time = time.time()
    while time.time() - last_update_time < max_idle_time:
        try:
            chunk = session.read_eager()
            if chunk:
                response += chunk
                last_update_time = time.time()  # 重置空闲计时
            else:
                time.sleep(check_interval)
        except telnetlib.TelnetTimeoutError:
            break
    return response

# 使用示例
session = telnetlib.Telnet(host, port, timeout=5)
session.write(command + b"\n")
full_response = get_full_telnet_response(session)
print(full_response.decode())

2. 基于响应结束标记判断 #

如果目标设备的响应有固定结束标记（比如示例中响应末尾的in Hz\n），可以循环读取直到捕获到该标记：

import time
import telnetlib

def read_until_marker(session, end_marker, timeout=5):
    response = b""
    start_time = time.time()
    while time.time() - start_time < timeout:
        chunk = session.read_some()
        if not chunk:
            time.sleep(0.05)
            continue
        response += chunk
        if end_marker in response:
            return response
    return response

# 使用示例（以示例响应的末尾标记为例）
full_response = read_until_marker(session, b"in Hz\n")
print(full_response.decode())

二、更易用的Telnet替代模块 #

1. telnetlib3 #

支持异步IO，API设计更简洁，原生支持非阻塞读取，适配Python 3.7及以上版本：

import asyncio
import telnetlib3

async def telnet_interaction(host, port, command):
    reader, writer = await telnetlib3.open_connection(host, port)
    writer.write(command + "\n")
    await writer.drain()
    # 根据实际响应结束标记调整readuntil的参数
    response = await reader.readuntil(b"in Hz\n")
    print(response.decode())
    writer.close()
    await writer.wait_closed()

asyncio.run(telnet_interaction(host, port, command))

2. paramiko #

虽主打SSH协议，但也支持Telnet，API封装更稳定完善，适合对连接可靠性要求较高的场景：

import paramiko

telnet_conn = paramiko.Telnet(host, port)
telnet_conn.write(command + b"\n")
# 设置超时并读取到指定结束标记
full_response = telnet_conn.read_until(b"in Hz\n", timeout=5)
print(full_response.decode())
telnet_conn.close()

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