如何从aiohttp中的子进程中读取？

Question

我有一个非常简单的服务器，它从子进程读取并将数据传递给任何打开的 websockets。我运行遇到的问题是我从子进程读取的方法似乎以一种我似乎无法遵循的方式破坏了 aiohttp：

#!/usr/bin/env python3

import asyncio
from aiohttp import web
import subprocess

async def websocket_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    request.app['websockets'].append(ws)
    try:
        async for msg in ws:
            print(msg)
            await asyncio.sleep(1)
    finally:
        request.app['websockets'].remove(ws)
    return ws


async def on_shutdown(app):
    for ws in app['websockets']:
        await ws.close(code=999, message='Server shutdown')

这是出错的地方：

async def listen_to_process(app):
    print("listen to process")
    while True:
        print(" looping? ")
        await asyncio.sleep(0.1)
        # the problem seems to be here
        line = await app['process'].stdout.readline()
        # if line:
        #     buffer.append(line)

async def start_background_tasks(app):
    app['process_listener'] = app.loop.create_task(listen_to_process(app))


async def cleanup_background_tasks(app):
    print('cleanup background tasks...')
    app['process_listener'].cancel()
    await app['process_listener']


def init():
    app = web.Application()
    app['websockets'] = []
    app.router.add_get('/updates', websocket_handler)
    cmd = "very long running subprocess"
    app['process'] = subprocess.Popen(cmd.split(" "), stdout=subprocess.PIPE)
    app.on_startup.append(start_background_tasks)
    app.on_cleanup.append(cleanup_background_tasks)
    app.on_shutdown.append(on_shutdown)
    return app


web.run_app(init())

所以，我的问题是：如何在我的应用程序后台循环中从 stdout 读取行？非常感谢您的指点。

Answer 1

请使用asyncio subprocess API:

app['process'] = await asyncio.create_subprocess_exec(
                           shlex.split(cmd), stdout=subprocess.PIPE)

小巧：使用 shlex 模块将命令行安全地拆分为参数列表。

如何从aiohttp中的子进程中读取？

How to read from subprocess in aiohttp?

python

subprocess

aiohttp