为何用aiofile异步读取文件比同步读取慢15倍？

我在测试命名管道（named pipe）的异步读取方案时，发现文件读取速度远慢于预期。资料显示该问题不仅限于命名管道，普通文件异步读取也存在类似情况。我分别实现了基于aiofile的异步读取命名管道代码，以及基于Popen的同步读取代码，通过time命令测试发现，异步程序的运行速度仅为同步程序的1/15。虽然已知aiofile存在性能问题，但15倍的差距仍超出预期，想了解推荐的异步文件读取方法，尤其是针对命名管道的专用优化方案。

异步代码示例 #

import sys, os
from asyncio import create_subprocess_exec, gather, run
from asyncio.subprocess import DEVNULL
from aiofile import async_open

async def read_strace(namedpipe):
    with open("async.log", "w") as outfp:
        async with async_open(namedpipe, "r") as npfp:
            async for line in npfp:
                outfp.write(line)

async def main(cmd):
    try:
        myfifo = os.mkfifo('myfifo', 0o600)
        process = await create_subprocess_exec(
            "strace", "-o", "myfifo", *cmd, 
            stdout=DEVNULL, stderr=DEVNULL)
        await gather(read_strace("myfifo"), process.wait())
    finally:
        os.unlink("myfifo")

run(main(sys.argv[1:]))

同步代码示例 #

from subprocess import Popen, DEVNULL
import sys, os

def read_strace(namedpipe):
    with open("sync.log", "w") as outfp:
        with open(namedpipe, "r") as npfp:
            for line in npfp:
                outfp.write(line)
   
def main(cmd):
    try:
        myfifo = os.mkfifo('myfifo', 0o600)
        process = Popen(
            ["strace", "-o", "myfifo", *cmd],
            stdout=DEVNULL, stderr=DEVNULL)
        read_strace("myfifo"),
    finally:
        os.unlink("myfifo")

main(sys.argv[1:])

测试结果 #

$ time ./async_program.py  find .   
poetry run ./async_program.py find .  4.06s user 4.75s system 100% cpu 8.727 total
$ time ./sync_program.py find .
poetry run ./sync_program.py find .  0.27s user 0.07s system 76% cpu 0.438 total

优化方案 #

1. 用原生asyncio监听文件描述符（Linux/macOS） #

Linux和macOS的asyncio支持通过loop.add_reader直接监听文件描述符，结合批量读取减少系统调用，能大幅降低异步调度开销：

import sys, os
import asyncio
from asyncio import create_subprocess_exec, gather, run, get_running_loop
from asyncio.subprocess import DEVNULL

async def read_strace(namedpipe):
    with open("async_opt.log", "w") as outfp, open(namedpipe, "r") as npfp:
        loop = get_running_loop()
        fd = npfp.fileno()
        buffer = []

        def read_callback():
            try:
                # 批量读取4KB数据，减少系统调用次数
                data = os.read(fd, 4096)
                if not data:
                    loop.remove_reader(fd)
                    return
                buffer.append(data.decode())
            except Exception:
                loop.remove_reader(fd)

        loop.add_reader(fd, read_callback)
        # 等待读取事件结束
        while loop.get_reader(fd):
            await asyncio.sleep(0.001)
        outfp.write(''.join(buffer))

async def main(cmd):
    try:
        os.mkfifo('myfifo', 0o600)
        process = await create_subprocess_exec(
            "strace", "-o", "myfifo", *cmd, 
            stdout=DEVNULL, stderr=DEVNULL)
        await gather(read_strace("myfifo"), process.wait())
    finally:
        os.unlink("myfifo")

run(main(sys.argv[1:]))

2. 用StreamReader对接管道流 #

命名管道本质是双向流，可以通过os.pipe()创建管道后，让子进程输出到管道写入端，再用asyncio.StreamReader读取，避开aiofile的低效包装：

import sys, os
import asyncio
from asyncio import create_subprocess_exec, gather, run, StreamReader, StreamReaderProtocol
from asyncio.subprocess import DEVNULL

async def read_strace(reader):
    with open("async_stream.log", "w") as outfp:
        while True:
            # 批量读取数据
            data = await reader.read(4096)
            if not data:
                break
            outfp.write(data.decode())

async def main(cmd):
    read_fd, write_fd = os.pipe()
    try:
        process = await create_subprocess_exec(
            "strace", "-o", f"/proc/self/fd/{write_fd}", *cmd, 
            stdout=DEVNULL, stderr=DEVNULL, pass_fds=[write_fd])
        # 关联StreamReader到管道读取端
        loop = get_running_loop()
        reader = StreamReader()
        protocol = StreamReaderProtocol(reader)
        await loop.connect_read_pipe(lambda: protocol, os.fdopen(read_fd, "rb"))
        await gather(read_strace(reader), process.wait())
    finally:
        os.close(write_fd)

run(main(sys.argv[1:]))

3. 核心优化点：减少上下文切换 #

不管采用哪种异步方案，都要避免逐行异步读取（async for line）——这种方式会频繁触发异步调度上下文切换，累积开销极大。改用固定大小的批量读取（如4KB、8KB），能显著降低系统调用和调度的总开销。

性能差距的关键原因 #

• aiofile的逐行异步读取是纯Python实现，没有同步读取的C层缓冲区优化，每次行读取都要触发异步调度，开销远超同步操作。
• 命名管道本身是阻塞IO，异步框架对其的细粒度包装如果没有批量处理，反而会放大调度开销，抵消异步的优势。

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