asyncio异步socket服务端运行循环时无法接收指令中断问题

核心问题原因 #

• 阻塞式Socket调用：代码中使用的标准recv()是同步阻塞方法，运行时会卡住整个asyncio事件循环，无法调度其他异步任务
• 接收逻辑被主动阻塞：收到startLoop指令后直接await task2和await task3，会导致recv_loop函数暂停执行，直到两个任务全部跑完才会继续接收新的指令，因此endLoop指令无法被及时处理

修复后的完整代码 #

server.py #

# python 3.7+
import asyncio

class SocketHandler():
    def __init__(self, reader, writer):
        self.reader = reader
        self.writer = writer
        self.run_loop = False
        self.loop_tasks = []

    async def recv_loop(self):
        try:
            print('client connected')
            while True:
                cmd_bytes = await self.reader.read(1024)  # 使用asyncio异步读接口
                cmd = cmd_bytes.decode()
                print(cmd)
                if len(cmd) == 0:
                    break
                elif cmd == "startLoop":
                    self.run_loop = True
                    # 仅创建任务，不await，避免阻塞接收逻辑
                    task2 = asyncio.create_task(self.whileLoop())
                    task3 = asyncio.create_task(test_counter())
                    self.loop_tasks = [task2, task3]
                elif cmd == "endLoop":
                    self.run_loop = False
        finally:
            # 连接关闭前等待未完成任务
            if self.loop_tasks:
                await asyncio.gather(*self.loop_tasks, return_exceptions=True)
            self.writer.close()
            await self.writer.wait_closed()

    async def whileLoop(self):
        count = 0
        while self.run_loop:
            print('self.run_loop: ' + str(self.run_loop))
            await asyncio.sleep(1)
            count += 1
            if count > 5:
                break

async def test_counter():
    for k in range(5):
        print(str(k))
        await asyncio.sleep(1)

async def handle_client(reader, writer):
    handler = SocketHandler(reader, writer)
    await handler.recv_loop()

async def main():
    server_address = ("127.0.0.1", 22000)
    print('starting up on %s port %s' % server_address)
    # 直接使用asyncio自带的异步Socket服务，无需手动处理阻塞问题
    server = await asyncio.start_server(
        handle_client,
        *server_address,
        reuse_address=True
    )
    async with server:
        await server.serve_forever()

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

client.py（无需修改） #

import time
import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 
sock.connect(('127.0.0.1', 22000))

sock.sendall(b'startLoop')
time.sleep(2)
sock.sendall(b'endLoop')

sock.close()

修改点说明 #

• 替换原生socket为asyncio自带的start_server异步流接口，使用异步read()方法代替阻塞recv()，避免卡住事件循环
• 收到startLoop指令后仅创建异步任务存入列表，不做await操作，让recv_loop可以继续接收后续指令
• 收到endLoop后修改run_loop标记，whileLoop任务会在下一次循环检测到标记变化自动退出，符合预期运行逻辑

内容的提问来源于stack exchange，提问作者Lilliën