如何将Python过程式同步代码改造为异步代码实现协作式多任务？

你可以通过asyncio事件循环+主动协作式让出控制权的方式实现需求，无需全量改造主应用为异步，也不用线程/多进程，确保WebSocket消息只会在主应用当前函数执行完毕后再处理。

核心思路 #

1. 用asyncio的WebSocket客户端（推荐websockets库）实现后台监听器，收到消息后暂存在异步队列中，不立即处理。
2. 将主应用的同步逻辑拆分为多个执行单元，在每个单元执行完毕后，主动调用await asyncio.sleep(0)让出控制权，触发事件循环处理队列中的WebSocket消息。
3. 利用asyncio.gather同时启动WebSocket监听器和主应用逻辑，实现单线程内的协作式多任务。

具体实现步骤 #

1. 依赖安装 #

如果使用websockets库，先安装：

pip install websockets

2. 代码实现 #

import asyncio
import websockets
import time

# 异步队列：暂存WebSocket消息，确保主函数执行完再处理
message_queue = asyncio.Queue()

async def websocket_listener(ws_url):
    """WebSocket后台监听器：接收消息并放入队列"""
    async with websockets.connect(ws_url) as websocket:
        async for message in websocket:
            await message_queue.put(message)
            # async for本身是异步非阻塞的，无需额外让出

def sync_task_1():
    """原同步主应用任务1"""
    print("[主应用] 执行同步任务1")
    # 模拟耗时业务逻辑
    time.sleep(2)

def sync_task_2():
    """原同步主应用任务2"""
    print("[主应用] 执行同步任务2")
    time.sleep(1)

async def main_app():
    """包装后的主应用逻辑：主动插入协作点"""
    while True:
        # 完整执行第一个同步任务，不会被中断
        sync_task_1()
        # 主动让出控制权，让事件循环处理WebSocket消息
        await asyncio.sleep(0)
        # 处理队列中积累的所有消息
        while not message_queue.empty():
            msg = await message_queue.get()
            print(f"[WebSocket] 处理消息：{msg}")
        
        # 完整执行第二个同步任务
        sync_task_2()
        await asyncio.sleep(0)
        # 再次处理消息
        while not message_queue.empty():
            msg = await message_queue.get()
            print(f"[WebSocket] 处理消息：{msg}")

async def main():
    """启动事件循环，同时运行监听器和主应用"""
    await asyncio.gather(
        websocket_listener("ws://your-websocket-server-url"),
        main_app()
    )

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

关键说明 #

• 协作式上下文切换：await asyncio.sleep(0)是触发事件循环切换的核心，它会告知事件循环当前任务已完成一个阶段，可以去处理其他任务。这保证了主应用的每个同步任务都会完整执行完毕后，才会处理WebSocket消息，不会出现随机中断。
• 消息暂存：用asyncio.Queue暂存WebSocket消息，既避免消息丢失，又确保只有主应用准备好时才处理。
• 最小改造量：无需修改原同步函数的内部逻辑，只需在函数调用后添加协作点即可。如果主应用是一个长耗时的单一函数，可以在函数内部的安全阶段（比如循环迭代末尾）插入await asyncio.sleep(0)。

特殊场景处理 #

如果主应用存在无法拆分的长阻塞函数（比如无法插入协作点的计算密集型任务），可以考虑用asyncio.get_running_loop().run_in_executor(None, long_blocking_func)将其放到线程池中执行——这属于"尽量避免"而非完全禁止线程的范畴，主应用逻辑仍会在单独线程中完整执行，不会被WebSocket任务抢占。

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