Python多线程实现同时处理键盘输入与Socket连接的问题咨询

问题根因排查 #

• 1. 协程执行逻辑理解错误：await 作用是挂起当前所在协程，等待目标异步操作返回后，才会继续执行当前协程的后续代码。你将异步输入、socket监听逻辑放在同一个协程的循环里串行执行，代码运行到await ainput.asyncInput()时，当前唯一的主协程就会挂起等待键盘输入，没有输入就永远不会走到后续监听连接的逻辑，自然表现为卡住。
• 2. 使用了阻塞式socket：标准库socket默认是阻塞模式，accept()、recv()都是阻塞方法，一旦执行就会卡住整个asyncio事件循环，就算解决了输入的问题，后续监听连接时也会无法响应键盘输入。
• 3. 代码细节错误：conn.close 缺少括号，属于只引用方法没有调用，实际不会关闭连接。

修复方案 #

你需要把键盘输入监听和socket服务处理拆成两个独立的协程任务，交给asyncio事件循环并行调度，同时将socket改为asyncio原生的异步socket实现，避免阻塞事件循环。

第一步：优化异步输入代码（避免每次调用都重复创建线程池） #

# asyncron_Input.py
import asyncio
from concurrent.futures import ThreadPoolExecutor

# 全局只初始化一次单线程池
_executor = ThreadPoolExecutor(1, 'Async Input')
async def asyncInput():
    return await asyncio.get_event_loop().run_in_executor(_executor, input)

第二步：重写服务端主逻辑，拆分并行协程 #

import asyncio
import asyncron_Input as ainput

server_address = ('localhost', 6969)

async def handle_client(reader, writer):
    """异步处理单个客户端连接"""
    client_addr = writer.get_extra_info('peername')
    print(f"connected to {client_addr}")
    while True:
        data = await reader.read(16)
        if not data:
            print("no data, close connection")
            break
        print(f"received: {data}")
        writer.write(b"successful")
        await writer.drain()
    writer.close()
    await writer.wait_closed()

async def run_server():
    """异步运行socket监听服务"""
    server = await asyncio.start_server(handle_client, *server_address)
    print(f"starting up on {server_address[0]} port {server_address[1]}")
    print("End server with 1")
    async with server:
        await server.serve_forever()

async def monitor_input():
    """监听键盘输入，输入1则关闭服务"""
    while True:
        i = await ainput.asyncInput()
        if i == "1":
            # 取消所有运行中的任务触发退出
            for task in asyncio.all_tasks():
                task.cancel()
            break

async def main():
    # 同时启动两个异步任务并行调度
    input_task = asyncio.create_task(monitor_input())
    server_task = asyncio.create_task(run_server())
    try:
        await asyncio.gather(input_task, server_task)
    except asyncio.CancelledError:
        print("server closed")

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

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