使用 asyncio 逐行读取文件

我希望在写入多个日志文件时读取它们并使用 asyncio 处理它们的输入。该代码必须在 Windows 上运行。根据我在 stackoverflow 和网络上的搜索了解到，异步文件 I/O 在大多数操作系统上都很棘手（select例如，将无法按预期工作）。虽然我确信我可以使用其他方法（例如线程）来做到这一点，但我想尝试一下 asyncio 看看它是什么样的。最有帮助的答案可能是描述该问题的解决方案的“架构”应该是什么样子的，即应该如何调用或调度不同的函数和协程。

下面给了我一个逐行读取文件的生成器（通过轮询，这是可以接受的）：

import time

def line_reader(f):
    while True:
        line = f.readline()
        if not line:
            time.sleep(POLL_INTERVAL)
            continue
        process_line(line)

由于需要监视和处理多个文件，此类代码需要线程。我对其进行了稍微修改，使其更适合 asyncio 使用：

import asyncio

def line_reader(f):
    while True:
        line = f.readline()
        if not line:
            yield from asyncio.sleep(POLL_INTERVAL)
            continue
        process_line(line)

当我通过异步事件循环安排它时，这种方法有效，但是如果process_data块，那当然不好。刚开始时，我想象解决方案会是这样的

def process_data():
    ...
    while True:
        ...
        line = yield from line_reader()
        ...

但我不知道如何做到这一点（至少在没有process_data管理相当多的状态）。

关于如何构建此类代码有什么想法吗？

使用aiofiles https://github.com/Tinche/aiofiles:

async with aiofiles.open('filename', mode='r') as f:
    async for line in f:
        print(line)

EDIT 1

正如 @Jashandeep 提到的，你应该关心阻塞操作：

另一种方法是select and or epoll:

from select import select

files_to_read, files_to_write, exceptions = select([f1, f2], [f1, f2], [f1, f2], timeout=.1)

The timeout参数在这里很重要。

see: https://docs.python.org/3/library/select.html#select.select https://docs.python.org/3/library/select.html#select.select

EDIT 2

您可以使用以下方法注册文件以进行读/写：循环.add_reader() https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.AbstractEventLoop.add_reader

它在循环内使用内部 EPOLL 处理程序。

EDIT 3

但请记住，Epoll 不适用于常规文件。