Прочитать файл по строкам с помощью asyncio

Я хочу прочитать несколько файлов журнала, когда они написаны, и обрабатывать их ввод с помощью asyncio. Код должен запускаться в окнах. Из того, что я понимаю из поиска как stackoverflow, так и Интернета, асинхронный ввод-вывод файлов сложен для большинства операционных систем (select не будет работать, как предполагалось, например). Хотя я уверен, что могу сделать это с помощью других методов (например, потоков), я бы попробовал асинхронно, чтобы понять, что это такое. Наиболее полезным ответом, вероятно, будет тот, который описывает, как должна выглядеть "архитектура" решения этой проблемы, т.е. Как вызывать или планировать различные функции и сопрограммы.

Ниже приведен генератор, который читает файлы по строкам (через опрос, который является приемлемым):

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)

Этот вид работает, когда я планирую его через цикл событий asyncio, но если process_data блокирует, то это, конечно, не очень хорошо. Когда я начинал, я думал, что решение будет выглядеть примерно так:

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

но я не мог понять, как сделать эту работу (по крайней мере, не без process_data управления довольно немного состояния).

Любые идеи о том, как я должен структурировать такой код?

Ответ 1

Из того, что я понимаю из поиска как в stackoverflow, так и в Интернете, асинхронный ввод-вывод файлов сложен для большинства операционных систем (выбор не будет работать, как предполагалось, например). Хотя я уверен, что могу сделать это с помощью других методов (например, потоков), я бы попробовал асинхронно, чтобы понять, что это такое.

asyncio является select на основе * nix систем под капотом, поэтому вы не сможете делать неблокирующий файловый ввод-вывод без использования потоков. В Windows asyncio может использовать IOCP, который поддерживает неблокирующий ввод-вывод файлов, но это не поддерживается asyncio.

Ваш код в порядке, за исключением того, что вы должны блокировать вызовы ввода-вывода в потоках, чтобы вы не блокировали цикл событий, если ввод-вывод работает медленно. К счастью, очень просто отключить загрузку работы с потоками, используя функцию loop.run_in_executor.

Сначала настройте выделенный пул потоков для ввода/вывода:

from concurrent.futures import ThreadPoolExecutor
io_pool_exc = ThreadPoolExecutor()

И затем просто выгрузите любые блокирующие вызовы ввода-вывода исполнителю:

...
line = yield from loop.run_in_executor(io_pool_exc, f.readline)
...

Ответ 2

Используя aiofiles:

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

РЕДАКТИРОВАТЬ 1

Как упоминалось в @Jashandeep, вы должны заботиться об операциях блокировки:

Другой метод: select и epoll:

from select import select

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

Здесь важен параметр timeout.

см. https://docs.python.org/2/library/select.html#select.select

РЕДАКТИРОВАТЬ 2

Вы можете зарегистрировать файл для чтения/записи с помощью: loop.add_reader()

Ответ 3

Ваша структура кода выглядит хорошо для меня, следующий код отлично работает на моей машине:

import asyncio

PERIOD = 0.5

@asyncio.coroutine
def readline(f):
    while True:
        data = f.readline()
        if data:
            return data
        yield from asyncio.sleep(PERIOD)

@asyncio.coroutine
def test():
    with open('test.txt') as f:
        while True:
            line = yield from readline(f)
            print('Got: {!r}'.format(line))

loop = asyncio.get_event_loop()
loop.run_until_complete(test())

Ответ 4

asyncio пока не поддерживает файловые операции, извините.

Таким образом, это не может помочь с вашей проблемой.