5

В своих пауках обычно использую такую конструкцию:

async def crawl(future, client, pool):
    futures = []
    # Получаем из футуры ссылки
    urls = await future
    # Выгребаем для каждой ссылки разметку страницы
    for request_future in asyncio.as_completed([request(client, url) for url in urls]):
        # Передаём парсинг разметки в пул потоков
        parse_future = loop.run_in_executor(pool, parse, (await request_future))
        # Рекурсивно вызываем себя для парсинга новой порции ссылок
        futures.append(asyncio.ensure_future(crawl(parse_future, client, pool)))
    # Это нужно только для того, чтобы знать
    # когда завершать цикл событий
    if futures:
        await asyncio.wait(futures)

Все хорошо бы, но когда имеется большое количество ссылок, то после выполнения запроса к серверу (request), результат сохраняется, и потом будет обрабатываться пулом потоком.

Вот именно когда складируется, потребляется большое количество оперативной памяти (вот сейчас почти 8 Гб).

Можно ли как нибудь изменить данную конструкцию, чтобы после запроса, результат сразу обрабатывался, ну либо сохранялся до определенного количества (например 50), а потом пулом потоков быстро обрабатывался.

5
  • По видимому проблема в цикле for. Он начинает работать только тогда, когда полностью формируется множество asyncio.as_completed([request(client, url) for url in urls]) 15 окт 2019 в 6:14
  • Всё таки for url in urls - это надо как то переделать имхо. Цикл должен быть привязан к url-ам, а не к завершенным потокам... Например использовать функцию с yield 15 окт 2019 в 6:19
  • stackoverflow.com/questions/35699601/… - не поможет? Вроде много вкусного. 15 окт 2019 в 6:26
  • Как вариант кстати - использовать совмещенную функцию с парсингом сразу вместо request(client, url) 15 окт 2019 в 6:29
  • @МихаилАлексеевич отмечаю на последний комментарий, какой смысл тогда использовать асинхронные запросы, если они у нас вместе с парсингом будут идти
    – danilshik
    15 окт 2019 в 9:49

2 ответа 2

3

В решении с чанками есть 2 проблемы:

  • дождаться выполнения чанки целиком: если попадается один тормозной запрос - он тормозит всю очередь.
  • рекурсия делает новые чанки: при большой вложенности одновременных запросов может быть экспонециально больше 50

Получаются превышения производительности и простои, при том не компенсирующие друг друга.

Для равномерной обработки запросов лучше использовать asyncio.Semaphore(value=50)

async def parse(data, req_sema):
    urls = getturls(data)
    crawl(urls, req_sema)

async def request(url, req_sema):
    print(f'url {url} запрошен')
    async with req_sema:
        print(f'url {url} запущен')
        await asyncio.sleep(1)
        print(f'url {url} завершен')


async def crawl(urls, req_sema):
    for request_future in asyncio.as_completed([request(url, req_sema) for url in urls]):
        data = await request_future
        await parse(data, req_sema)


async def main():
    req_sema = asyncio.Semaphore(value=50)
    urls = range(101)
    await crawl(urls,req_sema)

Начальное значение 50 значит что семафор можно захватить 50 раз и это гарантирует только 50 одновременных запросов. Объявленный один раз он действует на любую вложенность.

Дальше есть момент с обработкой в экзекуторе. Если она быстрая, то достаточно семафоров в запросах. Если медленная, то async with req_sema: меняем на раздельные захват - освобождение:

async def request(url, req_sema):
    await sem.acquire()
    await asyncio.sleep(1)


async def crawl(urls, req_sema):
    for request_future in asyncio.as_completed([request(url, req_sema) for url in urls]):
        await request_future
        urls = await parse()
        req_sema.release()
        await crawl(urls, req_sema)

Ограничение на рекурсию. Функция которая может захватить семафор не должна вызываться пока семафор захвачен. То есть crawl вызываете после release или вне with

2
  • у меня метод parse может возвращать ссылки, которые тоже нужно обрабатывать, а вы получается сломали сам метод crawl (паук)
    – danilshik
    20 дек 2019 в 14:12
  • да, вы можете вернуть рекурсию в код. я не стал её тут указывать потому как не несет смысловой нагрузки и только перегружают пример. передавайте семофор в каждый вызов или объявите его глобально - он также будет блокировать и рекурсивные вызовы. сейчас обновлю пример.
    – eri
    20 дек 2019 в 14:21
0

Придумал способ, решил делить все приходящие ссылки на пакеты по N-размеру (например 50). Получается мы делаем 50 запросов, и потом сразу 50 этих ответов от запросов парсим пулом потоков.

Получается мы не ожидаем завершения всех запросов (при большом количестве ссылок), тем самым не потребляем оперативную память на хранение результатов запросов, пока не закончим отправлять запросы, чтобы перейти к разбору пулом потоков

async def crawl(future, client, pool):
    futures = []
    # Получаем из футуры ссылки
    urls = await future
    new_urls = chunks(urls, limit_packet)

    for new_url in new_urls:
        for request_future in asyncio.as_completed([request(client, url) for url in new_url]):
            # Передаём парсинг разметки в пул потоков
            parse_future = loop.run_in_executor(pool, parse, (await request_future))
            # parse_future = loop.run_in_executor(pool, parse, (await request_future))
            # Рекурсивно вызываем себя для парсинга новой порции ссылок
            futures.append(asyncio.ensure_future(crawl(parse_future, client, pool)))
        # Это нужно только для того, чтобы знать
        # когда завершать цикл событий
        if futures:
            await asyncio.wait(futures)

def chunks(lst, count):
    new_list = []
    div, mod = divmod(len(lst), count)
    #Целое
    for i in range(div):
        temp = []
        for j in range(count):
            temp.append(lst[i * count + j])
        new_list.append(temp)
    #Остаток
    temp = []
    for i in range(mod):
        temp.append(lst[div * count + i])
    new_list.append(temp)
    return new_list

Быстро написал функция для разбиения списка на списки списков. Например при указании limit_packet = 50, потребление оперативной памяти Python не составило 200 МБайт

В теории такой метод будет работать немного медленнее, но он хотябы оперативы не жрет.

Если есть у кого более оригинальные идеи, будет интересно посмотреть

1
  • 1
    попробуй на семафорах сделать. в этом случае ждешь завершение блока. на семафорах можно поддерживать 50 воркеров не нарезая блоки.
    – eri
    9 ноя 2019 в 9:50

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими условиями использования и подтверждаете, что прочитали и поняли наши политику конфиденциальности и нормы поведения.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.