contextmanager中的Python Asyncio - Eventloop

Question

由于各种原因，我不喜欢使用loop.run()的方法，所以我想编写上下文循环，因为docs在不同的场合声明，如果不使用规范的.run()，就必须自己防止内存泄漏(i.e)。经过一番研究，似乎python用“我们不需要它”来回答这个特性！虽然如果使用异步的较低级别的api，contextmanagers总体上似乎非常好，但请参见PEP 343 -“with”语句 exampel 10：

这可以用来决定性地用close方法关闭任何东西，不管是文件、生成器还是其他什么东西。它甚至可以在对象不能保证要求关闭时使用(例如，接受任意迭代的函数)。

，所以无论如何我们能做吗？

相关链接：

• https://bugs.python.org/issue24795
• https://bugs.python.org/issue32875
• https://groups.google.com/g/python-tulip/c/8bRLexUzeU4
• https://bugs.python.org/issue19860#msg205062
• https://github.com/python/asyncio/issues/261

Answer 1

是的，我们可以为我们的事件循环设置一个上下文管理器，即使在c++实现(即)的情况下，通过子类似乎没有很好的实践。基本上，下面的构思如下：

TL;DR

1. 使用__enter__和__exit__创建一个对象，使语法with工作。
2. 与通常返回对象不同，我们返回异步服务的循环。
3. 包装asycio.loop.close()，以便循环被停止，我们的__exit__方法在前面被调用。
4. 关闭所有可能导致内存泄漏的连接，然后关闭循环。

​

侧记-注意

实现应该是一个包装器对象，它将一个新循环返回到一个匿名块语句中。请注意loop.stop() 将完成循环，不应调用进一步的操作。。总的来说，在我看来，下面的代码只是一个小小的帮助和更多的样式选择，特别是因为它不是真正的子类。但是我认为，如果有人想要使用更低的api而不考虑在之前完成所有的事情，那么这里有一种可能性。

import asyncio

class OpenLoop:

    def close(self,*args, **kwargs):
        self._loop.stop()

    def _close_wrapper(self):
        self._close = self._loop.close
        self._loop.close = self.close

    def __enter__(self):
        self._loop = asyncio.new_event_loop()
        self._close_wrapper()
        return self._loop
    
    def __exit__(self,*exc_info):
        asyncio.run(self._loop.shutdown_asyncgens())
        asyncio.run(self._loop.shutdown_default_executor())
        #close other services
        self._close()


if __name__ == '__main__':
    with OpenLoop() as loop:
        loop.call_later(1,loop.close)
        loop.run_forever()
    assert loop.is_closed()