从concurrent.futures到异步

Question

我对concurrent.futures有两个问题

如何在python concurrent.futures中破坏concurrent.futures()？

结论: time.sleep()不能中断。一个解决方案是:你可以在它周围写一个循环，然后做短暂的睡眠。

请参阅如何在python concurrent.futures中拆分concurrent.futures()

concurrent.futures的个人超时？

结论:个人超时需要由用户实现。例如:对于每次超时，您都可以调用等()。

请参阅concurrent.futures的个别超时

问题

异步能解决这些问题吗？

Answer 1

在异步模型中，执行由事件循环进行调度和协调。要取消当前挂起的任务的执行，您基本上不需要继续执行它。虽然这在实践中有点不同，但很明显，这使得取消暂停的任务在理论上很简单。

当然，单独的超时也是有可能的:每当您挂起一个coroutine以等待结果时，您都希望提供一个超时值。当到达超时且任务尚未完成时，事件循环将确保取消等待任务。

一些具体样本：

>>> import asyncio
>>> loop = asyncio.get_event_loop()
>>> task = asyncio.ensure_future(asyncio.sleep(5))
>>> task.cancel()
>>> loop.run_until_complete(task)
Traceback (most recent call last):
   ...
concurrent.futures._base.CancelledError

在实践中，可以使用这样的方法来实现：

class Foo:
    task = None

    async def sleeper(self):
        self.task = asyncio.sleep(60)
        try:
            await self.task
        except concurrent.futures.CancelledError:
            raise NotImplementedError

当该方法处于休眠状态时，其他人可以调用foo.task.cancel()来唤醒协同线并让它处理取消操作。或者，无论谁打电话给sleeper()，都可以直接取消它，而不给它一个清理的机会。

设置超时也同样容易：

>>> loop.run_until_complete(asyncio.wait_for(asyncio.sleep(60), 5))
[ ... 5 seconds later ... ]
Traceback (most recent call last):
   ...
concurrent.futures._base.TimeoutError

特别是在HTTP请求超时的上下文中，请参见艾奥赫特

async def fetch_page(session, url):
    with aiohttp.Timeout(10):
        async with session.get(url) as response:
            assert response.status == 200
            return await response.read()

with aiohttp.ClientSession(loop=loop) as session:
    content = loop.run_until_complete(fetch_page(session, 'http://python.org'))

显然，每次对fetch_page的调用都可以决定自己的aiohttp.Timeout值，当达到超时时，每个单独的实例都会抛出自己的异常。

Answer 2

您可以在它的异常中立即引发(使用asyncio.CancelledError)。

为了克服这个问题，我使用了这种方法：

import asyncio

async def worker():
    try:
        # await for some coroutine process
    except asyncio.CancelledError:
        # Do stuff
        raise asyncio.CancelledError()
    except Exception as exc:
        # Do stuff
        print(exc)
    finally:
        await asyncio.sleep(2)