Python asyncio 不显示任何错误

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

1. 填写一个Queue使用一个单一的一次性处理一堆网址Producer
2. 产生一堆Consumers
3. Each Consumer保持异步地从 中提取 urlQueue并发送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))

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

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

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

另一种选择是未处理的要传播到的消费者异常run。这必须显式安排，但具有永远不允许异常静默通过的优点。 （看本文以获得该主题的详细处理。）实现它的一种方法是等待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

问题：如何检测和处理asyncio中的异常？

异常通过以下方式传播await通常会像其他代码一样进行检测和处理。仅需要特殊处理来捕获从“后台”任务泄漏的异常，例如consumer.

如何在不中断队列的情况下重试？

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