Python asyncio 不显示任何错误

Question

我正在尝试使用 asyncio 从数千个网址中获取一些数据。以下是设计的简要概述:

1. 填写Queue使用单个 Producer 一次性处理一堆网址
2. 生成一堆Consumers
3. 每个Consumer不断从Queue异步提取url并发送GET请求
4. 对结果进行一些后处理
5. 合并所有处理结果并返回

问题: asyncio几乎从不显示是否有任何问题，它只是静静地挂起，没有错误。我把print到处声明自己发现问题，但没有多大帮助。

根据输入网址的数量以及消费者或限制的数量，我可能会收到以下错误:

1. Task was destroyed but it is pending!
2. task exception was never retrieved future: <Task finished coro=<consumer()
3. aiohttp.client_exceptions.ServerDisconnectedError
4. aiohttp.client_exceptions.ClientOSError: [WinError 10053] An established connection was aborted by the software in your host machine

问题:如何检测和处理 asyncio 中的异常？如何在不中断 Queue 的情况下重试？

贝娄是我编译的代码，查看了异步代码的各种示例。目前，def get_video_title 末尾存在故意错误。功能。运行时，什么也没有显示。

import asyncio
import aiohttp
import json
import re
import nest_asyncio
nest_asyncio.apply() # jupyter notebook throws errors without this


user_agent = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36"

def get_video_title(data):
    match = re.search(r'window\[["\']ytInitialPlayerResponse["\']\]\s*=\s*(.*)', data)
    string = match[1].strip()[:-1]
    result = json.loads(string)
    return result['videoDetails']['TEST_ERROR'] # <---- should be 'title'

async def fetch(session, url, c):
    async with session.get(url, headers={"user-agent": user_agent}, raise_for_status=True, timeout=60) as r:
        print('---------Fetching', c)
        if r.status != 200:
            r.raise_for_status()
        return await r.text()

async def consumer(queue, session, responses):
    while True:
        try:
            i, url = await queue.get()
            print("Fetching from a queue", i)
            html_page = await fetch(session, url, i)

            print('+++Processing', i)
            result = get_video_title(html_page) # should raise an error here!
            responses.append(result)
            queue.task_done()

            print('+++Task Done', i)

        except (aiohttp.http_exceptions.HttpProcessingError, asyncio.TimeoutError) as e:
            print('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>Error', i, type(e))
            await asyncio.sleep(1)
            queue.task_done()

async def produce(queue, urls):
    for i, url in enumerate(urls):
        print('Putting in a queue', i)
        await queue.put((i, url))

async def run(session, urls, consumer_num):
    queue, responses = asyncio.Queue(maxsize=2000), []

    print('[Making Consumers]')
    consumers = [asyncio.ensure_future(
        consumer(queue, session, responses)) 
                 for _ in range(consumer_num)]

    print('[Making Producer]')
    producer = await produce(queue=queue, urls=urls)

    print('[Joining queue]')
    await queue.join()

    print('[Cancelling]')
    for consumer_future in consumers:
        consumer_future.cancel()

    print('[Returning results]')
    return responses

async def main(loop, urls):
    print('Starting a Session')
    async with aiohttp.ClientSession(loop=loop, connector=aiohttp.TCPConnector(limit=300)) as session:
        print('Calling main function')
        posts = await run(session, urls, 100)
        print('Done')
        return posts


if __name__ == '__main__':
    urls = ['https://www.youtube.com/watch?v=dNQs_Bef_V8'] * 100
    loop = asyncio.get_event_loop()
    results = loop.run_until_complete(main(loop, urls))

Answer 1

问题是您的消费者仅捕获两个非常具体的异常，并在它们的情况下将任务标记为已完成。如果发生任何其他异常，例如与网络相关的异常，它将终止消费者。但是，run 无法检测到这一点，它正在等待 queue.join()，而消费者(有效地)在后台运行。这就是您的程序挂起的原因 - 排队的项目永远不会被考虑，并且队列永远不会被完全处理。

有两种方法可以解决此问题，具体取决于您希望程序在遇到意外异常时执行的操作。如果您希望它继续运行，您可以向消费者添加一个包罗万象的 except 子句，例如:

        except Exception as e
            print('other error', e)
            queue.task_done()

另一种方法是将未处理的消费者异常传播到run。这必须显式安排，但具有永远不允许异常静默通过的优点。 (有关该主题的详细处理，请参阅 this article。)实现它的一种方法是同时等待 queue.join() 和消费者；由于消费者处于无限循环中，因此只有在发生异常时才会完成。

    print('[Joining queue]')
    # wait for either `queue.join()` to complete or a consumer to raise
    done, _ = await asyncio.wait([queue.join(), *consumers],
                                 return_when=asyncio.FIRST_COMPLETED)
    consumers_raised = set(done) & set(consumers)
    if consumers_raised:
        await consumers_raised.pop()  # propagate the exception

Questions: how to detect and handle exceptions in asyncio?

异常通过await传播，并且通常像在任何其他代码中一样被检测和处理。仅需要特殊处理来捕获从“后台”任务(例如消费者)泄漏的异常。

how to retry without disrupting the Queue ?

您可以在 except block 中调用 await queue.put((i, url)) 。该项目将被添加到队列的后面，由消费者拾取。在这种情况下，您只需要第一个代码段，并且不想费心尝试将 consumer 中的异常传播到 run。