在 Python3 asyncio 中与子进程通信

Question

我有一个基于 asyncio 的程序（称之为服务器）。要求是从该服务器中启动其他进程，从该进程获取响应（这些是长时间运行的）并进一步处理/发送整理的响应。

请参阅下面的带有 cmets 的骨架代码，了解我想要实现的目标：

import asyncio
import subprocess

loop = asyncio.get_event_loop()


def logFileProcessorBgThread(fName):
    print("Starting logFileProcessorBgThread")
    f = subprocess.Popen(['tail','-f', LOG_FILE], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    while True:
        line = f.stdout.readline().decode().strip()
        print(line)
        # somehow, send back the line to the start_server


async def start_server():
    # I want to start the logFileProcessorBgThread as a separate process,
    # one for each input file
    files = ["/var/log/syslog", "/tmp/somefile.log"]
    for f in files:
        # start logFileProcessorBgThread(f) as a process

    # need to build a way to receive the updates from the 2 process as an
    # when they have data..

    # my requirement is to bring back the responses to this process, so I can
    # further handle them (in my case send it out on a secure channel that
    # this process has established. Other launched process would not have
    # access to this channel..)



def main():
    loop.run_until_complete(start_server())

if __name__ == '__main__':
    main()

我不清楚启动过程中的这种流式响应将如何工作。

Answer 1

您可以通过在事件循环中结合使用ProcessPoolExecutor 和run_in_executor(..) 函数来做到这一点。这是您示例的基本调整：

import asyncio
import os
import time
from concurrent.futures import ProcessPoolExecutor

loop = asyncio.get_event_loop()


def process(filename):
    # do some work
    print("{} from process {}".format(filename, os.getpid()))
    time.sleep(2)
    return "response {}".format(filename)

async def start_server():
    # create the process pool executors with 2 workers (# of processes to use)
    pool = ProcessPoolExecutor(2)
    filenames = ["/var/log/syslog", "/tmp/somefile.log"]

    # launch them in the pool and then await on the results
    futures = [loop.run_in_executor(pool, process, filename) for filename in filenames]
    for future in futures:
        response = await future
        print("response was '{}'".format(response))


def main():
    loop.run_until_complete(start_server())


if __name__ == '__main__':
    main()