python - 按顺序使用 Python 的 asyncio 获取数据

Question

我有一个 Python 2.7 程序，它从网站上提取数据并将结果转储到数据库中。它遵循消费者生产者模型，使用线程模块编写。

只是为了好玩，我想使用新的 asyncio 模块(从 3.4 开始)重写这个程序，但我不知道如何正确地做到这一点。

最关键的要求是程序必须按顺序从同一网站获取数据。例如对于 url ' http://a-restaurant.com ' 它应该首先得到 ' http://a-restaurant.com/menu/0 '，然后是 ' http://a-restaurant.com/menu/1 ', 然后 ' http://a-restaurant.com/menu/2 ', ...如果未获取它们，网站将完全停止传送页面，您必须从 0 开始。

然而，另一个网站 (' http://another-restaurant.com ') 的另一个抓取可以(并且应该)同时运行(其他网站也有顺序限制)。

threading 模块非常适合这个，因为我可以为每个网站创建单独的线程，并且在每个线程中它可以等到一个页面完成加载后再获取另一个页面。

下面是线程版本 (Python 2.7) 中的一个大大简化的代码片段:

class FetchThread(threading.Threading)
    def __init__(self, queue, url)
        self.queue = queue
        self.baseurl = url
    ...
    def run(self)
        # Get 10 menu pages in a sequantial order
        for food in range(10):
            url = self.baseurl + '/' + str(food)
            text = urllib2.urlopen(url).read()
            self.queue.put(text)
            ...
def main()
    queue = Queue.Queue()
    urls = ('http://a-restaurant.com/menu', 'http://another-restaurant.com/menu')
    for url in urls:
        fetcher = FetchThread(queue, url)
        fetcher.start()
        ...

下面是我尝试使用 asyncio(在 3.4.1 中)执行此操作的方式:

@asyncio.coroutine
def fetch(url):
    response = yield from aiohttp.request('GET', url)
    response = yield from response.read_and_close()
    return response.decode('utf-8')

@asyncio.coroutine
def print_page(url):
    page = yield from fetch(url)
    print(page)


l = []
urls = ('http://a-restaurant.com/menu', 'http://another-restaurant.com/menu')
for url in urls:
    for food in range(10):
        menu_url = url + '/' + str(food)
        l.append(print_page(menu_url))

loop.run_until_complete(asyncio.wait(l))

它以非连续的顺序获取和打印所有内容。好吧，我想这就是那些协程的全部想法。我不应该使用 aiohttp 而只使用 urllib 获取吗？但是第一家餐厅的抓取是否会阻止其他餐厅的抓取？我只是认为这完全错误吗？(这只是一个尝试按顺序获取东西的测试。还没有到达队列部分。)

Answer 1

您当前的代码适用于不关心请求顺序排序的餐厅。菜单的所有十个请求将同时运行，并在完成后立即打印到标准输出。

显然，这不适用于需要顺序请求的餐厅。您需要稍微重构一下才能使其正常工作:

@asyncio.coroutine
def fetch(url):
    response = yield from aiohttp.request('GET', url)
    response = yield from response.read_and_close()
    return response.decode('utf-8')

@asyncio.coroutine
def print_page(url):
    page = yield from fetch(url)
    print(page)

@syncio.coroutine
def print_pages_sequential(url, num_pages):
    for food in range(num_pages):
        menu_url = url + '/' + str(food)
        yield from print_page(menu_url)

l = [print_pages_sequential('http://a-restaurant.com/menu', 10)]

conc_url = 'http://another-restaurant.com/menu'
for food in range(10):
    menu_url = conc_url + '/' + str(food)
    l.append(print_page(menu_url))

loop.run_until_complete(asyncio.wait(l))

我们没有将顺序餐厅的所有十个请求都添加到列表中，而是将一个协程添加到列表中，它将按顺序遍历所有十个页面。它的工作方式是 yield from print_page 将停止执行 print_pages_sequential 直到 print_page 请求完成，但它不会阻塞任何其他并发运行的协程(就像您附加到 l 的所有 print_page 调用)。

通过这种方式，您所有的“另一家餐厅”请求都可以完全并发运行，就像您想要的那样，并且您的“一家餐厅”请求将按顺序运行，但不会阻塞任何“另一家餐厅” "请求。

编辑:

如果所有的站点都有相同的顺序抓取需求，逻辑可以进一步简化:

l = []
urls = ["http://a-restaurant.com/menu", "http://another-restaurant.com/menu"]
for url in urls:
    menu_url = url + '/' + str(food)
    l.append(print_page_sequential(menu_url, 10))

loop.run_until_complete(asyncio.wait(l))