如何将Python异步方法封装为普通方法?含Telnet3改造及替代库问询
改造异步Telnet3类为同步调用
方案:用asyncio.run封装异步方法
将原有的异步方法改为私有内部实现,对外提供同步接口,通过asyncio.run()执行异步逻辑。同时注意把异步代码中的time.sleep替换为asyncio.sleep,避免阻塞事件循环:
import asyncio import telnetlib3 class Telnet3: def __init__(self, host, port): self.host = host self.port = port self.reader = None self.writer = None def connect(self): # 同步接口,调用异步内部方法 return asyncio.run(self._connect()) async def _connect(self): self.reader, self.writer = await telnetlib3.open_connection(self.host, self.port) data = await asyncio.wait_for(self.reader.read(4096), timeout=2) return data def write(self, command): # 同步接口,调用异步内部方法 return asyncio.run(self._write(command)) async def _write(self, command): if not self.writer: raise ConnectionError("未连接到Telnet服务器") self.writer.write(command + "\r\n") await self.writer.drain() # 确保数据完全发送 await asyncio.sleep(5) # 替换time.sleep为异步睡眠,避免阻塞事件循环 data = await asyncio.wait_for(self.reader.read(4096), timeout=5) return data
这样你就可以按照预期的同步方式调用:
from mytelnetlib3 import Telnet3 mytelnet = Telnet3("localhost", 9000) mytelnet.connect() reply = mytelnet.write("some command") print(reply)
其他可用的Python Telnet库
标准库telnetlib(推荐)
Python自带的同步Telnet库,无需额外安装,直接支持同步调用,完全符合你的需求。实现类似封装类的示例:
import telnetlib import time class TelnetSync: def __init__(self, host, port): self.host = host self.port = port self.tn = None def connect(self): self.tn = telnetlib.Telnet(self.host, self.port, timeout=2) # 读取初始响应并解码为字符串 return self.tn.read_very_eager().decode('utf-8') def write(self, command): if not self.tn: raise ConnectionError("未连接到Telnet服务器") # 发送命令(需编码为字节) self.tn.write((command + "\r\n").encode('utf-8')) time.sleep(5) # 读取响应并解码为字符串 return self.tn.read_very_eager().decode('utf-8')
telnetlib2
一个增强版的Telnet库,兼容标准库telnetlib的API,增加了一些额外功能,同样是同步实现,可通过pip install telnetlib2安装。
Exscript实现方案(可选)
如果你需要使用Exscript,以下是符合需求的封装类示例(需先通过pip install exscript安装):
from Exscript.protocols import Telnet class ExscriptTelnet: def __init__(self, host, port): self.host = host self.port = port self.conn = None def connect(self): self.conn = Telnet() self.conn.connect(self.host, port=self.port, timeout=2) # 读取初始响应 return self.conn.receive() def write(self, command): if not self.conn: raise ConnectionError("未连接到Telnet服务器") self.conn.send(command + "\r\n") self.conn.sleep(5) return self.conn.receive()
内容的提问来源于stack exchange,提问作者Alex
相关产品推荐
相关产品推荐

