Python中的异步异常处理答案

【问题标题】：Asynchronous exception handling in PythonPython中的异步异常处理
【发布时间】：2015-05-20 23:04:46
【问题描述】：

我有以下代码使用asyncio 和aiohttp 发出异步HTTP 请求。

import sys
import asyncio
import aiohttp

@asyncio.coroutine
def get(url):
    try:
        print('GET %s' % url)
        resp = yield from aiohttp.request('GET', url)
    except Exception as e:
        raise Exception("%s has error '%s'" % (url, e))
    else:
        if resp.status >= 400:
            raise Exception("%s has error '%s: %s'" % (url, resp.status, resp.reason))

    return (yield from resp.text())

@asyncio.coroutine
def fill_data(run):
    url = 'http://www.google.com/%s' % run['name']
    run['data'] = yield from get(url)

def get_runs():
    runs = [ {'name': 'one'}, {'name': 'two'} ]
    loop = asyncio.get_event_loop()
    task = asyncio.wait([fill_data(r) for r in runs])
    loop.run_until_complete(task)   
    return runs

try:
    get_runs()
except Exception as e:
    print(repr(e))
    sys.exit(1)

由于某种原因，get 函数内部引发的异常未被捕获：

Future/Task exception was never retrieved
Traceback (most recent call last):
  File "site-packages/asyncio/tasks.py", line 236, in _step
    result = coro.send(value)
  File "mwe.py", line 25, in fill_data
    run['data'] = yield from get(url)
  File "mwe.py", line 17, in get
    raise Exception("%s has error '%s: %s'" % (url, resp.status, resp.reason))
Exception: http://www.google.com/two has error '404: Not Found'

那么，处理协程引发的异常的正确方法是什么？

【问题讨论】：

标签： python python-3.x python-asyncio

【解决方案1】：

asyncio.wait 实际上并不消耗传递给它的Futures，它只是等待它们完成，然后返回Future 对象：

协程 asyncio.wait(futures, *, loop=None, timeout=None, return_when=ALL_COMPLETED)

等待 Futures 和协程对象由futures给出的序列来完成。协程将被包装在任务中。返回两组Future：（完成，待处理）。

直到您真正yield from done 列表中的项目，它们将保持未使用状态。由于您的程序在没有消耗期货的情况下退出，您会看到“从未检索到异常”消息。

对于您的用例，使用asyncio.gather 可能更有意义，它实际上会消耗每个Future，然后返回一个聚合所有结果的Future（或引发第一个Exception由输入列表中的未来抛出）。

def get_runs():
    runs = [ {'name': 'one'}, {'name': 'two'} ]
    loop = asyncio.get_event_loop()
    tasks = asyncio.gather(*[fill_data(r) for r in runs])
    loop.run_until_complete(tasks)
    return runs

输出：

GET http://www.google.com/two
GET http://www.google.com/one
Exception("http://www.google.com/one has error '404: Not Found'",)

请注意，asyncio.gather 实际上允许您在其中一个期货引发异常时自定义其行为；默认行为是引发它遇到的第一个异常，但它也可以只返回输出列表中的每个异常对象：

asyncio.gather(*coros_or_futures, loop=None, return_exceptions=False)

从给定的协程对象返回一个未来的聚合结果或期货。

所有期货必须共享同一个事件循环。如果所有的任务都完成了成功，返回的未来的结果是结果列表（在原始序列的顺序，不一定是结果到达）。 如果return_exceptions 是True，则任务被视为成功的结果，并收集在结果列表；否则，第一个引发的异常将立即传播到返回的未来。

【讨论】：

感谢您的解释，文档对异常处理并不十分清楚
那么如何使用wait 呢？是不是类似于yield from asyncio.wait(...)？ await asyncio.wait(...) 也应该工作吗？
顺便说一句，如果其中一个 couroutine 从未真正启动，则会出现同样的问题。对我来说，两个协程打开了一些套接字（手动）并尝试await <loop>.sock_recv(<socket>, <size>)。如果套接字未切换为非阻塞（使用<socket>.setblocking()），则不会启动第二个协程并且KeyboardInterrupt 导致“从未检索到任务异常”。
在异步函数内部，不能使用yield from 来解决未来或任务。一个应该await 代替。这对我来说不是很清楚。
感谢您提供return_exceptions 提示。执行数百个头部请求，我不希望抛出异常，因为其中一个超时。

【解决方案2】：

调试或“处理”callback 中的异常：

返回某些结果或引发异常的协程：

@asyncio.coroutine
def async_something_entry_point(self):
    try:
        return self.real_stuff_which_throw_exceptions()
    except:
        raise Exception(some_identifier_here + ' ' + traceback.format_exc())

还有回调：

def callback(self, future: asyncio.Future):
    exc = future.exception()
    if exc:
        # Handle wonderful empty TimeoutError exception
        if type(exc) == TimeoutError:
            self.logger('<Some id here> callback exception TimeoutError')
        else:
            self.logger("<Some id here> callback exception " + str(exc))

    # store your result where you want
    self.result.append(
        future.result()
    )

【讨论】：