有没有办法在Python中指定Replcio REPL的事件循环类型?

s4chpxco  于 6个月前  发布在  Python
关注(0)|答案(2)|浏览(79)

我正在用Python编写一些使用Pythoncio模块的网络代码,我喜欢使用-m Pythoncio中的Python REPL来测试示例。我注意到Pythoncio REPL的默认事件循环类型会根据操作系统而变化。例如,在Linux上,它使用PythonEvent Loop,而在Windows上,它使用Proactor。
我的问题是:
1.有没有什么方法可以指定REPL使用的事件循环(而不必自己打补丁,这显然是有效的)。
1.接下来的问题--如果没有的话--有没有一种方法可以用另一种类型的事件循环来替换线程中正在运行的事件循环?
让我知道你的想法。

uyto3xhc

uyto3xhc1#

是的,您是正确的。xmlcio模块使用的默认事件循环可能取决于操作系统。通常会选择最适合该平台的默认事件循环。
在Linux上,默认事件循环通常是SelectorEventLoop,在Windows上,默认事件循环是ProactorEventLoop。这是因为这两个平台的底层I/O模型不同。SelectorEventLoop基于select系统调用,非常适合Linux,而ProactorEventLoop基于Windows Proactor I/O模型。
如果你想显式设置事件循环类型,你可以在代码中这样做。例如,如果你总是想使用SelectorEventLoop,你可以显式设置它:

import asyncio

asyncio.set_event_loop(asyncio.SelectorEventLoop())

字符串
或者,如果你想检查当前正在使用哪个事件循环,你可以这样做:

import asyncio

loop = asyncio.get_event_loop()
print(f"Current event loop: {type(loop)}")

mutmk8jj

mutmk8jj2#

如果将来有人从Google来这里:我最终只是在asyncio.__main__模块中使用了现有的python c REPL代码,并为我自己的项目修改了它,这样我的项目就可以像python -m project一样调用,它会生成一个shell(根据我的偏好)。
这并不是一件很难的事情。但是我将向你展示如何修改REPLs环境,以便它也能导入你的项目:

class SelectorEventPolicy(asyncio.DefaultEventLoopPolicy):
    @staticmethod
    def exception_handler(self, context):
        print("exception handler")
        print(context)

    @staticmethod
    def loop_setup(loop):
        loop.set_debug(False)
        loop.set_exception_handler(SelectorEventPolicy.exception_handler)
        loop.default_exception_handler = SelectorEventPolicy.exception_handler

    def new_event_loop(self):
        selector = selectors.SelectSelector()
        loop = asyncio.SelectorEventLoop(selector)
        SelectorEventPolicy.loop_setup(loop)
        return loop

asyncio.set_event_loop_policy(SelectorEventPolicy())
class REPLThread(threading.Thread):
    def run(self):
        try:
            # ... previous code to push some text to the user
            # This code executes the statement in the new prompt
            # Modify it to suite your project
            console.push("from p2pd.do_imports import *")
            # ...

字符串
Python的Repll的原始模块在这里:https://github.com/python/cpython/blob/main/Lib/asyncio/main.py
而我自己的模块的fork(只有最小的修改)在这里:https://github.com/robertsdotpm/p2pd/blob/test_sprint1/p2pd/main.py
实际上是一样的,但我想在所有平台上使用选择器事件循环。可能有更好的方法来做到这一点,但这似乎对我的项目现在工作得很好。如果其他人从谷歌来到这里,也许有用。

相关问题