是否可以设置Python asyncio REPL中的事件循环类型？

Question

我正在使用Python的asyncio模块编写一些网络代码，并且非常喜欢通过-m asyncio命令行参数启动REPL来测试示例。然而，我发现asyncio REPL默认使用的事件循环类型会根据操作系统不同而变化。例如，在Linux系统中它使用Selector事件循环，而在Windows系统中则使用Proactor。

我的问题是：

是否有一种方法可以在不自行修改模块的情况下指定asyncio REPL所使用的事件循环类型？
如果无法直接指定，是否存在一种方法可以在运行中的线程中替换现有的事件循环为另一种类型的事件循环？

请分享一下您的想法。

Lukas Kalbertodt · Answer

您说得对，《asyncio》模块所使用的默认事件循环确实可能因操作系统而异，默认情况下通常会选择最适合该平台的事件循环类型。

在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"当前事件循环类型: {type(loop)}")

在Linux系统中默认情况下，这将输出：

（注：请使用Windows平台的用户补充在此环境下运行上述代码的输出结果）

Sam · Answer

如果未来有人从谷歌搜索来到这里：我最终选择使用了asyncio.__main__模块中已有的异步REPL代码，并将其修改以适应我的项目，使得我的项目可以通过类似python -m project的方式调用，并启动一个带有个人偏好设置的交互式shell。

这不是一件特别困难的事情。但我将展示如何修改REPL的环境以便同时导入你的项目：

class SelectorEventPolicy(asyncio.DefaultEventLoopPolicy):
    @staticmethod
    def exception_handler(self, context):
        print("异常处理器")
        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:
            # ... 前面的代码用于向用户推送一些提示信息
            # 这段代码执行新的提示符下的语句
            # 根据你的项目需求修改这部分
            console.push("from p2pd.do_imports import *")
            # ...

Python内置的异步REPL模块源码位于：
https://github.com/python/cpython/blob/main/Lib/asyncio/main.py

而我对该模块做了少量修改后用于自己项目的版本在这里：
https://github.com/robertsdotpm/p2pd/blob/test_sprint1/p2pd/main.py

两者基本上是相同的，但我希望在所有平台上都使用selector事件循环。可能有更优的方法实现这一点，但目前这种方法在我的项目中运行良好。对于其他通过谷歌搜索来到这里的开发者而言，或许这个方法也会有所帮助。