Как запускать задачи одновременно в asyncio? - PullRequest
Новая
       27

Как запускать задачи одновременно в asyncio?

0 голосов
/ 12 января 2019

Я пытаюсь научиться запускать задачи одновременно с помощью модуля Python asyncio. В следующем коде у меня есть макет "веб-сканер" для примера. По сути, я пытаюсь сделать это там, где в любой момент времени происходит максимум два активных запроса fetch (), и я хочу, чтобы process () вызывался в течение периода sleep (). 

import asyncio

class Crawler():

    urlq = ['http://www.google.com', 'http://www.yahoo.com', 
            'http://www.cnn.com', 'http://www.gamespot.com', 
            'http://www.facebook.com', 'http://www.evergreen.edu']

    htmlq = []
    MAX_ACTIVE_FETCHES = 2
    active_fetches = 0

    def __init__(self):
        pass

    async def fetch(self, url):
        self.active_fetches += 1
        print("Fetching URL: " + url);
        await(asyncio.sleep(2))
        self.active_fetches -= 1
        self.htmlq.append(url)

    async def crawl(self):
        while self.active_fetches < self.MAX_ACTIVE_FETCHES:
            if self.urlq:
                url = self.urlq.pop()
                task = asyncio.create_task(self.fetch(url))
                await task
            else:
                print("URL queue empty")
                break;

    def process(self, page):
        print("processed page: " + page)

# main loop

c = Crawler()
while(c.urlq):
    asyncio.run(c.crawl())
    while c.htmlq:
        page = c.htmlq.pop()
        c.process(page)

Однако приведенный выше код загружает URL-адреса один за другим (а не два одновременно) и не выполняет никакой «обработки» до тех пор, пока не будут получены все URL-адреса. Как сделать так, чтобы задачи fetch () выполнялись одновременно и чтобы между ними во время сна () вызывался процесс ()?

Ответы [ 2 ]

0 голосов
/ 12 января 2019

Вы можете сделать htmlq asyncio.Queue() и изменить htmlq.append на htmlq.push. Тогда ваш main может быть асинхронным, например: 

async def main():
    c = Crawler()
    asyncio.create_task(c.crawl())
    while True:
        page = await c.htmlq.get()
        if page is None:
            break
        c.process(page)

Ваш код верхнего уровня сводится к вызову asyncio.run(main()).

Как только вы закончите сканирование, crawl() может поставить в очередь None, чтобы уведомить главную сопрограмму о том, что работа выполнена.

0 голосов
/ 12 января 2019

Ваш crawl метод ожидает после каждого отдельного задания; Вы должны изменить это на это: 

async def crawl(self):
    tasks = []
    while self.active_fetches < self.MAX_ACTIVE_FETCHES:
        if self.urlq:
            url = self.urlq.pop()
            tasks.append(asyncio.create_task(self.fetch(url)))
    await asyncio.gather(*tasks)

РЕДАКТИРОВАТЬ : Вот более чистая версия с комментариями, которая извлекает и обрабатывает все одновременно, сохраняя при этом базовую возможность ограничить максимальное количество сборщиков. 

import asyncio

class Crawler:

    def __init__(self, urls, max_workers=2):
        self.urls = urls
        # create a queue that only allows a maximum of two items
        self.fetching = asyncio.Queue()
        self.max_workers = max_workers

    async def crawl(self):
        # DON'T await here; start consuming things out of the queue, and
        # meanwhile execution of this function continues. We'll start two
        # coroutines for fetching and two coroutines for processing.
        all_the_coros = asyncio.gather(
            *[self._worker(i) for i in range(self.max_workers)])

        # place all URLs on the queue
        for url in self.urls:
            await self.fetching.put(url)

        # now put a bunch of `None`'s in the queue as signals to the workers
        # that there are no more items in the queue.
        for _ in range(self.max_workers):
            await self.fetching.put(None)

        # now make sure everything is done
        await all_the_coros

    async def _worker(self, i):
        while True:
            url = await self.fetching.get()
            if url is None:
                # this coroutine is done; simply return to exit
                return

            print(f'Fetch worker {i} is fetching a URL: {url}')
            page = await self.fetch(url)
            self.process(page)

    async def fetch(self, url):
        print("Fetching URL: " + url);
        await asyncio.sleep(2)
        return f"the contents of {url}"

    def process(self, page):
        print("processed page: " + page)


# main loop
c = Crawler(['http://www.google.com', 'http://www.yahoo.com', 
             'http://www.cnn.com', 'http://www.gamespot.com', 
             'http://www.facebook.com', 'http://www.evergreen.edu'])
asyncio.run(c.crawl())
...