利用 asyncio 生成器和 asyncio.as_completed

Question

我有一些代码用于抓取 url，解析信息，然后使用 SQLAlchemy 将其放入数据库。我正在尝试异步执行此操作，同时限制同时请求的最大数量。

这是我的代码：

async def get_url(aiohttp_session, url1, url2):
    async with session.get(url1) as r_url1:
       if r_url1.status == 200:
          async with session.get(url2) as r_url2:
             if r_url2.status == 200:
                return await r_url1.json(), await r_url2.json()

async def url_generator(formatted_start_date, formatted_end_date, machine_id, interval):
    interval_start = formatted_start_date
    interval_end = formatted_start_date + interval

    while interval_end <= formatted_end_date:
        yield (f"https://example.org/start={interval_start}"
               f"Start={datetime.strftime(interval_start, DATETIME_FORMAT)}"
               f"&End={datetime.strftime(interval_end, DATETIME_FORMAT)}"
               f"&machines={machine_id}",
               f"https://example.org/start={interval_start}"
               f"Start={datetime.strftime(interval_start, DATETIME_FORMAT)}"
               f"&End={datetime.strftime(interval_end, DATETIME_FORMAT)}"
               f"&machines={machine_id}&groupby=Job"
               )
        interval_start += interval
        interval_end += interval

async def parse(database, url1_json, url2_json):
    """ Do some parsing and save it using credentials stored in the database object """


def main(database, formatted_start_date, formatted_end_date, machine_id, interval):
    async for url1_json, url2_json in asyncio.as_completed(url_generator(formatted_start_date, formatted_end_date, machine_id, interval)):
         parse(database, url1_json, url2_json)

我收到错误yield from should be used as context manager expression。

我已经尝试查看 documentation here 以及同步原语，但仍然对我出错的地方以及我应该如何从我的生成器创建任务感到困惑。

Answer 1

贴出的代码有几个问题：

• 您正在尝试将as_completed 用作异步迭代器，并使用async for 对其结果进行迭代。但是，as_completed 不会返回异步迭代器（至少 not yet），必须使用常规的 for 进行迭代，并像 shown in the docs 一样显式等待每个生成的对象。

• 您将异步迭代器传递给as_completed，而它接受普通容器或（常规）可迭代器。

• 您在未使用async def 定义的函数中使用async for，这应该是语法错误。另外，parse() 被定义为协程，你不用等待它。

好消息是，由于url_generator 已经是一个生成器，所以您根本不需要as_completed，您应该能够对其进行迭代：

async def main(database, formatted_start_date, formatted_end_date,
               machine_id, interval):
    async for url1_json, url2_json in url_generator(
            formatted_start_date, formatted_end_date,
            machine_id, interval)):
        await parse(database, url1_json, url2_json)

但是请注意，async for 不会自动并行化迭代，它只会允许其他协程与迭代的协程并行运行。要并行化迭代，您需要调用create_task 来并行提交任务，并使用asyncio.Semaphore 来限制并行任务的数量。例如：

async def parse(database, url1_json, url2_json, limit):
    # async with applied to a semaphore ensures that no more than N
    # coroutines that use the same semaphore enter the "with" block
    # in parallel
    async with limit:
        ... code goes here ...

async def main(database, formatted_start_date, formatted_end_date,
               machine_id, interval):
    limit = asyncio.Semaphore(10)

    # create all coroutines in advance using create_task
    # and run them in parallel, relying on the semaphore
    # limit the number of simultaneous requests
    tasks = []
    async for url1_json, url2_json in url_generator(
            formatted_start_date, formatted_end_date,
            machine_id, interval)):
        # this create_task just creates the task - it will
        # start running when we return to the event loop
        tasks.append(asyncio.create_task(parse(database, url1_json, url2_json, limit))

    # suspend to the event loop, resuming this coroutine only after
    # all the tasks have finished (or any of them raises)
    await asyncio.gather(*tasks)

请注意，url_generator 不需要异步，因为它不需要 await 任何东西。您可以使用def 定义它并使用for 对其进行迭代。