在两个子进程之间使用管道的异步流处理数据

Question

您好，我在这里找不到示例的解决方案。我找到了https://stackoverflow.com/a/36666420/6089311，但是没有流读。

我想运行两个独立于 python 程序的子进程（在后台运行）。第一个子进程通过管道为第二个进程提供数据，我想通过流的方式对 stdout 行进行一些处理。

下面的例子被屏蔽了，但我不知道为什么：

import asyncio
import os

async def foo():
    read, write = os.pipe()
    process_1 = await asyncio.create_subprocess_exec('ls', stdout=write)
    process_2 = await asyncio.create_subprocess_exec('wc', stdin=read, stdout=asyncio.subprocess.PIPE)
    
    async for l in process_2.stdout:
        # streaming process data
        print(l)
    
    os.close(write)
    os.close(read)

await foo() # jupyter call
# async.run(foo()) # python call

Answer 1

如果我将 close() 移动到您的链接等位置，代码对我有用。

但这可能不是你所期望的。

import asyncio
import os

async def foo():
    read, write = os.pipe()
    
    process_1 = await asyncio.create_subprocess_exec('ls', stdout=write)
    os.close(write)

    process_2 = await asyncio.create_subprocess_exec('wc', stdin=read, stdout=asyncio.subprocess.PIPE)
    os.close(read)
    
    async for line in process_2.stdout:
        # streaming process data
        print(line.decode())

#await foo() # jupyter call
asyncio.run(foo()) # python call

---

最终我可以稍后关闭read，但我必须在for-loop 之前关闭write。

import asyncio
import os

async def foo():
    read, write = os.pipe()
    
    process_1 = await asyncio.create_subprocess_exec('ls', stdout=write)
    process_2 = await asyncio.create_subprocess_exec('wc', stdin=read, stdout=asyncio.subprocess.PIPE)
    
    os.close(write)

    async for line in process_2.stdout:
        # streaming process data
        print(line.decode())

    os.close(read)

#await foo() # jupyter call
asyncio.run(foo()) # python call