You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何将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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.18 06:32:09