如何替换asyncio.get_event_loop()以避免DeprecationWarning？

你遇到的警告是因为Python 3.10+对asyncio.get_event_loop()的行为做了调整——当没有当前运行的事件循环时，这个方法不再自动创建新循环，而是抛出弃用警告，原来的写法已经不符合新版本规范了。你之前直接替换成new_event_loop()没用，是因为没有把新创建的循环设置为当前上下文的事件循环。

下面给你两种可行的修复方案：

方案一：把_run_queue改成异步函数（推荐） #

这是最符合asyncio设计理念的做法，直接用asyncio.run()来启动异步任务，它会自动帮你创建、运行并关闭事件循环：

async def flush(self, queue_c) -> None:
    try:
        loop = asyncio.get_running_loop()
        result_objects_action = loop.run_in_executor(None, self.create_objects, queue_c)
        # do stuff
    except:
        # except handling

async def _run_queue(self) -> None:
    # 简化列表生成，替代原while循环
    coroutine_list = [self.flush(c) for c in range(len(self._queue))]
    await asyncio.gather(*coroutine_list)
    self._queue = []

# 调用_run_queue的地方改成这样：
# asyncio.run(self._run_queue())

把_run_queue改成异步函数后，用await替代run_until_complete，外部调用时用asyncio.run()启动即可，全程不需要手动管理事件循环，彻底解决警告问题。

方案二：保持_run_queue为同步函数（兼容旧写法） #

如果因为某种原因必须让_run_queue保持同步，那需要手动创建事件循环并设置为当前循环：

async def flush(self, queue_c) -> None:
    try:
        loop = asyncio.get_running_loop()
        result_objects_action = loop.run_in_executor(None, self.create_objects, queue_c)
        # do stuff
    except:
        # except handling

def _run_queue(self) -> None:
    coroutine_list = [self.flush(c) for c in range(len(self._queue))]
    # 手动创建并设置当前事件循环
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(asyncio.gather(*coroutine_list))
    finally:
        loop.close()  # 用完记得关闭循环避免资源泄漏
    self._queue = []

先创建新循环，用set_event_loop把它设为当前上下文的循环，之后再调用run_until_complete，用完后关闭循环即可消除警告。

内容的提问来源于stack exchange，提问作者Bob van Luijt