python asyncio- contextmanager中的eventloop

发布于 2025-02-12 04:55:34 字数 1333 浏览 0 评论 0 原文

由于我不喜欢使用 loop.run（）的方法，因此我想编码上下文循环，因为文档在不同的情况下指出，如果您不使用Canonical .run（）您必须自己防止内存泄漏（ ie ）。经过一番研究，似乎Python Devs用回答了此功能，我们不需要它！。虽然如果您使用asyncio的较低级别的API，则一般看起来似乎完全可以，请参见 Exampel 10：

这可用于确定性地关闭任何内容方法，无论是文件，生成器还是其他方法。甚至可以使用当不能保证对象需要关闭时（例如，功能接受任意迭代）

那么我们还是可以做到吗？

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

情仇皆在手 2025-02-19 04:55:36

是的，即使由于C实现（ie）。基本上，下面制定的想法如下：

tl; dr

用 __ enter __ 和 __ exit __ 创建一个对象/代码>工作。
我们不通常返回对象，而是返回由asyncio包装
asycio.loop.close（）的循环，以便停止循环，我们的 __ exit __ exit __ exit __ 方法获取获取被调用之前。
关闭所有可能导致内存泄漏的连接，然后关闭循环。

旁注

实现是由于包装对象，该对象将 new 循环返回到匿名块语句中。请注意， 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()

Yes we can have a context manager for our event loop, even if there seems no good practice via subclassing due the c implementations (i.e). Basically the idea crafted out below is the following:

TL;DR

Create an Object with __enter__ and __exit__ to have the syntax with working.
Instead of usually returning the object, we return the loop that is served by asyncio
wrapping the asycio.loop.close() so that the loop gets stopped and our __exit__ method gets invoked before.
Close all connections which can lead to memory leaks and then close the loop.

Side note

The implementation is due a wrapper object that returns a new loop into an anonymous block statement. Be aware that loop.stop() will finalize the loop and no further actions should be called. Overall the code below is just a little help and more a styling choice in my opinion, especially due the fact that it is no real subclass. But I think if someone wants to use the lower api without minding to finalize everything before, here is a possibility.

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()

回复收藏 0 原文

~没有更多了~