从concurrent.futures到asyncio

Question

我有两个与concurrent.futures相关的问题：

如何在python concurrent.futures中打破time.sleep()？

结论：time.sleep()无法被中断。一种解决方法是：可以在它周围编写循环并进行短暂的休眠。

参见如何打破python concurrent.futures中的time.sleep()

为concurrent.futures设置单独的超时时间？

结论：需要用户实现单独的超时时间。例如：对于每个超时，您可以调用wait()。

参见为concurrent.futures设置单独的超时时间

问题

asyncio是否解决了这些问题？

- guettli

---

为了自包含起见，您能在这里总结一下那其他两个问题吗？ - deceze

Answer 1

在asyncio模型中，执行由事件循环调度和协调。要取消当前暂停的任务的执行，您基本上只需要不恢复它即可。虽然实践中会有一些差异，但理论上应该很明显，这使得取消暂停的任务变得简单。

同样，可以通过相同的方式设置每个单独的超时时间：每当您挂起协程以等待结果时，都需要提供超时值。事件循环将确保在达到超时时间并且任务尚未完成时取消等待任务。

以下是一些具体的示例：

>>> 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请求超时的情况下，参见aiohttp：

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的值，而每个独立的实例在达到超时时都会抛出自己的异常。

- deceze

---

哇，太棒了。谢谢你。想象一下我使用子进程模块（我看到它被支持了），然后我想以某种方式终止子进程。你有什么提示怎样做吗？ - guettli

不是的，你应该为此开一个新问题。 - deceze

请查看asyncio.subprocess模块以及这些示例。 - Vincent

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)

- Benyamin Jafari