Python asyncio 不显示任何错误

Question

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

1. Queue一次性填写一堆网址Producer
2. 产生一堆 Consumers
3. 每个都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？

Bellow 是我在查看异步代码的各种示例时编译的代码。目前，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

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

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

        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))在except块中调用。该项目将被添加到队列的后面，由消费者拿起。在这种情况下，您只需要第一个片段，并且不想费心尝试将异常传播consumer到run.