时间戳

首页 > 解决方案 > 获取嵌套网址时如何在异步中链接协程

python - 获取嵌套网址时如何在异步中链接协程

问题描述

我目前正在设计一个蜘蛛来抓取一个特定的网站。我可以做到同步，但我试图让我的头脑围绕 asyncio 使其尽可能高效。我已经尝试了很多不同的方法，yield但chained functions我queues无法使其工作。

我对解决问题的设计部分和逻辑最感兴趣。不需要可运行的代码，而是突出 assyncio 最重要的方面。我不能发布任何代码，因为我的尝试不值得分享。

使命：

example.com（我知道，应该是 example.com）的设计如下：

在此处输入图像描述

以同步方式，逻辑将是这样的：

for table in my_url_list:
    # Get HTML
    # Extract urls from HTML to user_list
    for user in user_list:
        # Get HTML
        # Extract urls from HTML to user_subcat_list
        for subcat in user_subcat_list:
            # extract content

但现在我想异步抓取网站。假设我们使用 5 个实例（pyppeteer 中的选项卡或 aiohttp 中的请求）来解析内容。我们应该如何设计它以使其最高效，我们应该使用什么 asyncio 语法？

更新

感谢@user4815162342 解决了我的问题。我一直在玩弄他的解决方案，如果其他人想玩 asyncio，我会在下面发布可运行的代码。

import asyncio
import random
 
my_url_list = ['exemple.com/table1', 'exemple.com/table2', 'exemple.com/table3']


# Random sleeps to simulate requests to the server
async def randsleep(caller=None):
    i = random.randint(1, 6)
    if caller:
        print(f"Request HTML for {caller} sleeping for {i} seconds.")
    await asyncio.sleep(i)


async def process_urls(url_list):
    print(f'async def process_urls: added {url_list}')
    limit = asyncio.Semaphore(5)
    coros = [process_user_list(table, limit) for table in url_list]
    await asyncio.gather(*coros)


async def process_user_list(table, limit):
    async with limit:
        # Simulate HTML request and extracting urls to populate user_list
        await randsleep(table)
        if table[-1] == '1':
            user_list = ['exemple.com/user1', 'exemple.com/user2', 'exemple.com/user3']
        elif table[-1] == '2':
            user_list = ['exemple.com/user4', 'exemple.com/user5', 'exemple.com/user6']
        else:
            user_list = ['exemple.com/user7', 'exemple.com/user8', 'exemple.com/user9']

        print(f'async def process_user_list: Extracted {user_list} from {table}')

    # Execute process_user in parallel, but do so outside the `async with`
    # because process_user will also need the semaphore, and we don't need
    # it any more since we're done with fetching HTML.
    coros = [process_user(user, limit) for user in user_list]
    await asyncio.gather(*coros)


async def process_user(user, limit):
    async with limit:
        # Simulate HTML request and extracting urls to populate user_subcat_list
        await randsleep(user)
        user_subcat_list = [user + '/profile', user + '/info', user + '/followers']
        print(f'async def process_user: Extracted {user_subcat_list} from {user}')

    coros = [process_subcat(subcat, limit) for subcat in user_subcat_list]
    await asyncio.gather(*coros)


async def process_subcat(subcat, limit):
    async with limit:
        # Simulate HTML request and extracting content
        await randsleep(subcat)
        print(f'async def process_subcat: Extracted content from {subcat}')

if __name__ == '__main__':
    asyncio.run(process_urls(my_url_list))

标签： pythonweb-scrapingpython-asynciopython-3.7

解决方案

让我们重新构建同步代码，以便可以访问网络的每个部分都在一个单独的函数中。功能没有改变，但它会让以后的事情变得更容易：

def process_urls(url_list):
    for table in url_list:
        process_user_list(table)

def process_user_list(table):
    # Get HTML, extract user_list
    for user in user_list:
        process_user(user)

def process_user(user):
    # Get HTML, extract user_subcat_list
    for subcat in user_subcat_list:
        process_subcat(subcat)

def process_subcat(subcat):
    # get HTML, extract content

if __name__ == '__main__':
    process_urls(my_url_list)

假设处理顺序无关紧要，我们希望异步版本for并行运行现在在循环中调用的所有函数。它们仍将在单个线程上运行，但它们会处理await任何可能阻塞的事情，从而允许事件循环并行化等待并通过在准备好继续时恢复每个协程来驱动它们完成。这是通过将每个协程生成为独立于其他任务并因此并行运行的单独任务来实现的。例如，顺序（但仍然是异步的）版本process_urls如下所示：

async def process_urls(url_list):
    for table in url_list:
        await process_user_list(table)

这是异步的，因为它在事件循环中运行，您可以并行运行多个这样的函数（我们将很快展示如何执行），但它也是顺序的，因为它选择await每次调用process_user_list. 在每次循环迭代中，await显式指示 asyncio 暂停执行 of，process_urls直到结果process_user_list可用为止。

相反，我们想要的是告诉 asyncioprocess_user_list并行运行所有调用，并暂停执行process_urls直到它们全部完成。在“后台”中生成协程的基本原语是将其调度为使用的任务asyncio.create_task，这是最接近轻量级线程的异步等效项。使用create_task并行版本process_urls看起来像这样：

async def process_urls(url_list):
    # spawn a task for each table
    tasks = []
    for table in url_list:
        asyncio.create_task(process_user_list(table))
        tasks.append(task)
    # The tasks are now all spawned, so awaiting one task lets
    # them all run.
    for task in tasks:
        await task

乍一看，第二个循环看起来像以前版本一样按顺序等待任务，但事实并非如此。由于每个都await暂停到事件循环，等待任何任务允许所有任务进行，只要它们事先使用create_task(). 无论他们完成的顺序如何，总等待时间都不会超过最长任务的时间。

这种模式经常被使用，以至于 asyncio 有一个专门的实用程序函数asyncio.gather。使用此函数，相同的代码可以用更短的版本表示：

async def process_urls(url_list):
    coros = [process_user_list(table) for table in url_list]
    await asyncio.gather(*coros)

但是还有另一件事需要注意：因为process_user_list将从服务器获取 HTML，并且将有许多并行运行的实例，我们不能允许它通过数百个同时连接来锤击服务器。我们可以创建一个工作任务池和某种队列，但是 asyncio 提供了一个更优雅的解决方案：信号量。Semaphore 是一种同步设备，它不允许超过预定数量的并行激活，使其余部分排队等待。

的最终版本process_urls会创建一个信号量并将其传递下去。它不会激活信号量，因为process_urls它本身实际上并没有获取任何 HTML，因此它没有理由在process_user_lists 运行时保持信号量槽。

async def process_urls(url_list):
    limit = asyncio.Semaphore(5)
    coros = [process_user_list(table, limit) for table in url_list]
    await asyncio.gather(*coros)

process_user_list看起来很相似，但它确实需要使用以下方法激活信号量async with：

async def process_user_list(table, limit):
    async with limit:
        # Get HTML using aiohttp, extract user_list

    # Execute process_user in parallel, but do so outside the `async with`
    # because process_user will also need the semaphore, and we don't need
    # it any more since we're done with fetching HTML.
    coros = [process_user(user, limit) for user in user_list]
    await asyncio.gather(*coros)

process_user并且process_subcat更多相同：

async def process_user(user, limit):
    async with limit:
        # Get HTML, extract user_subcat_list
    coros = [process_subcat(subcat, limit) for subcat in user_subcat_list]
    await asyncio.gather(*coros)

def process_subcat(subcat, limit):
    async with limit:
        # get HTML, extract content
    # do something with content

if __name__ == '__main__':
    asyncio.run(process_urls(my_url_list))

在实践中，您可能希望异步函数共享相同的 aiohttp 会话，因此您可能会在顶级函数（process_urls在您的情况下）中创建它并将其与信号量一起传递。每个获取 HTML 的函数都会有另一个async with用于 aiohttp 请求/响应的函数，例如：

async with limit:
    async with session.get(url, params...) as resp:
        # get HTML data here
        resp.raise_for_status()
        resp = await resp.read()
# extract content from HTML data here

两个async withs 可以合并为一个，减少缩进但保持相同的含义：

async with limit, session.get(url, params...) as resp:
    # get HTML data here
    resp.raise_for_status()
    resp = await resp.read()
# extract content from HTML data here

推荐阅读