python - asyncio.subprocess 总是阻塞

Question

我需要编写一些异步代码来运行子进程作为其任务的一部分。即使我使用 asyncio.subprocess 我的代码仍然阻塞。我的服务器看起来像这样:

import asyncio
import asyncio.subprocess
import websockets

async def handler(websocket, path):
    while True:
        data = await websocket.recv()
        print('I received a message')
        player = await asyncio.create_subprocess_exec(
            'sleep', '5',
            stdin=asyncio.subprocess.DEVNULL,
            stdout=asyncio.subprocess.DEVNULL,
            stderr=asyncio.subprocess.DEVNULL)
        await player.wait()
        print('Finished waiting')

server = websockets.serve(handler, '0.0.0.0', '8000')
asyncio.get_event_loop().run_until_complete(server)
asyncio.get_event_loop().run_forever()

还有一个非常基本的客户端:

导入异步导入网络套接字

async def client():
    async with websockets.connect('ws://localhost:8000') as websocket:
        for i in range(5):
            await websocket.send('message')
            await asyncio.sleep(0.5)

asyncio.get_event_loop().run_until_complete(client())

我希望输出看起来像这样:

I received a message
I received a message
I received a message
I received a message
I received a message
Finished waiting
Finished waiting
Finished waiting
Finished waiting
Finished waiting

但我得到的是:

I received a message
Finished waiting
I received a message
Finished waiting
I received a message
Finished waiting
I received a message
Finished waiting
I received a message
Finished waiting

每行“我收到一条消息”后等待 5 秒。

Answer 1

await player.wait() 行没有阻塞其他异步操作，而是等待 5 秒!

如果您不想等待响应，请尝试使用 ensure_future() 代替:

# add:
async def wait_for_player(player, path):
    print("Waiting...", path)
    await player.wait()
    print("Done", path)

# and replace await player.wait() with:
asyncio.ensure_future(wait_for_player(player, path))

实际上，您还可以将 create_subprocess_exec() 移动到 wait_for_player()。

---

要查看您的代码没有阻塞，请尝试以下操作:

客户:

import asyncio

import websockets


async def client(n):
    async with websockets.connect('ws://localhost:8000/{}/'.format(n)) as websocket:
        print(n, "start")
        for i in range(5):
            print(n, i)
            await websocket.send('message')
            await asyncio.sleep(0.5)
    print(n, "done")


tasks = [client(i) for i in range(5)]
asyncio.get_event_loop().run_until_complete(asyncio.wait(tasks))

服务器:

import asyncio
import asyncio.subprocess

import random
import websockets


async def handler(websocket, path):
    try:
        while True:
            data = await websocket.recv()
            pause = random.randint(1, 5)
            print('I received a message', path, "Pausing:", pause)
            player = await asyncio.create_subprocess_exec(
                'sleep', str(pause),
                stdin=asyncio.subprocess.DEVNULL,
                stdout=asyncio.subprocess.DEVNULL,
                stderr=asyncio.subprocess.DEVNULL)
            await player.wait()
            print('Finished waiting', path)
    except websockets.ConnectionClosed:
        print("Connection closed!", path)


server = websockets.serve(handler, '0.0.0.0', '8000')
asyncio.get_event_loop().run_until_complete(server)
asyncio.get_event_loop().run_forever()