AsyncIO Streams在FastAPI中正常运行但单独运行异常问题排查

核心问题分析 #

你的Sockets类和启动代码存在几个关键问题，直接导致了单独运行时的异常：

• 类属性与实例属性混淆：message_queue是类属性，所有Sockets实例会共享同一个队列，且read/write方法里直接引用message_queue而非self.message_queue，完全无法正确访问实例专属的队列。
• 低级笔误：write方法里的message.queue.pop(0)是明显的拼写错误，应该是self.message_queue.pop(0)。
• 事件循环生命周期问题：
  • 用asyncio.run(init())时，init函数执行完后，asyncio.run会自动关闭事件循环，导致start里创建的read/write后台任务被直接取消，程序立刻退出。
  • reader.read()返回空字节b''意味着Socket连接已被对方关闭，但你的代码没处理这个情况，会陷入无限循环读取空值。
  • wait_for的超时设置在单独运行时一旦触发，会直接抛出TimeoutError导致任务崩溃，而FastAPI/Uvicorn环境有全局异常捕获逻辑兜底。

修正后的完整代码 #

1. 修复后的Sockets类 #

# socket_service.py
import asyncio

class Sockets:
    def __init__(self, host: str, port: int):
        self.host = host
        self.port = port
        self.reader: asyncio.StreamReader | None = None
        self.writer: asyncio.StreamWriter | None = None
        # 改为实例属性，每个实例拥有独立消息队列
        self.message_queue = []

    async def start(self):
        try:
            reader, writer = await asyncio.wait_for(
                asyncio.open_connection(self.host, self.port),
                timeout=5
            )
            self.reader = reader
            self.writer = writer
            print(f"已连接到 {self.host}:{self.port}")
            # 无需手动获取事件循环，直接创建后台任务（Python3.7+支持）
            asyncio.create_task(self.read())
            asyncio.create_task(self.write())
        except asyncio.TimeoutError:
            print("连接超时")
            raise

    async def read(self):
        if not self.reader:
            return
        try:
            while True:
                # 改为读取固定长度（可根据实际协议调整），避免无限等待
                response = await asyncio.wait_for(
                    self.reader.read(4096),
                    timeout=60,
                )
                if not response:
                    print("连接已被对方关闭")
                    break
                print(f"收到消息: {response}")
                self.message_queue.append(response)
                # 移除不必要的sleep，避免消息处理延迟
                # await asyncio.sleep(1)
        except asyncio.TimeoutError:
            print("读取消息超时")
        finally:
            await self.close()

    async def write(self):
        if not self.writer:
            return
        try:
            while True:
                if self.message_queue:
                    msg = self.message_queue.pop(0)
                    self.writer.write(msg)
                    await self.writer.drain()
                # 短暂sleep避免空循环占用CPU
                await asyncio.sleep(0.1)
        except Exception as e:
            print(f"写入消息出错: {e}")
        finally:
            await self.close()

    async def close(self):
        if self.writer:
            self.writer.close()
            await self.writer.wait_closed()
        print("Socket连接已关闭")

2. 正确的单独启动代码 #

方式1：用asyncio.run保持事件循环运行

# manual_start.py
import asyncio
from socket_service import Sockets

async def init():
    # 替换为你的实际Socket服务地址和端口
    session = Sockets(host="127.0.0.1", port=8080)
    await session.start()
    # 用Event阻塞，让asyncio.run不会立即退出，保持后台任务运行
    await asyncio.Event().wait()

if __name__ == "__main__":
    try:
        asyncio.run(init())
    except KeyboardInterrupt:
        print("程序已手动终止")

方式2：手动管理事件循环

# manual_start.py
import asyncio
from socket_service import Sockets

async def init():
    session = Sockets(host="127.0.0.1", port=8080)
    await session.start()

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(init())
        loop.run_forever()
    except KeyboardInterrupt:
        print("程序已手动终止")
    finally:
        loop.close()

关键修正说明 #

1. 实例属性替代类属性：把message_queue移到__init__中初始化，每个实例拥有独立队列，所有方法统一用self.message_queue访问。
2. 修复笔误：修正write方法中的队列访问错误。
3. 处理连接关闭：当reader.read()返回空字节时，跳出循环并关闭连接，避免无限读取空值。
4. 保持事件循环运行：在init中用asyncio.Event().wait()阻塞，让asyncio.run不会在init执行完后立刻关闭事件循环。
5. 优化读取逻辑：用reader.read(4096)替代reader.read()，避免无限等待（可根据实际协议调整读取方式），移除不必要的sleep减少消息延迟。
6. 完善异常处理：添加连接超时、读取超时的捕获，以及关闭连接的finally逻辑，保证资源正确释放。

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