httpx客户端关闭触发RuntimeError: Event loop is closed问题修复咨询

问题场景 #

需要维护一个持久化的httpx AsyncClient以复用连接池，实现了单例模式的Request类，但运行时触发RuntimeError: Event loop is closed错误。仅取消注释nest_asyncio.apply()可解决，但不想依赖该停止维护的包。同时相同逻辑的aiohttp.ClientSession可正常运行。

实现代码 #

import asyncio
import weakref
from threading import Lock

import nest_asyncio
from httpx import AsyncClient


class Request:
    _instance = None
    _lock = Lock()
    _loop: asyncio.AbstractEventLoop = None
    _new_loop = False
    _session = None

    def __new__(cls):
        if not cls._instance:
            with cls._lock:
                if not cls._instance:
                    instance = super().__new__(cls)
                    cls._instance = instance
                    weakref.finalize(instance, instance._close)

                    if cls._loop is None:
                        try:
                            cls._loop = asyncio.get_running_loop()
                        except RuntimeError:
                            cls._loop = asyncio.new_event_loop()
                            cls._new_loop = True
                            asyncio.set_event_loop(cls._loop)

                if cls._new_loop:
                    # nest_asyncio.apply()
                    cls._loop.run_until_complete(cls.create_client())
                else:
                    cls._loop.create_task(cls.create_client())

        return cls._instance

    @classmethod
    async def create_client(cls):
        cls._session = AsyncClient()

    @classmethod
    def _close(cls):
        cls._loop.run_until_complete(cls.close())

        if cls._new_loop:
            cls._loop.close()

    @classmethod
    async def close(cls):
        if cls._session:
            await cls._session.aclose()

    async def get(self, url):
        try:
            return await self._session.get(url)

        except Exception:
            return None


req = Request()


async def main():
    result = await req.get('https://www.google.com')
    if result:
        print(result.text[:100])


asyncio.run(main())

错误信息 #

...
    await self._pool.aclose()
  File "/home/user/.cache/pypoetry/virtualenvs/app-gEFTwlce-py3.12/lib/python3.12/site-packages/httpcore/_async/connection_pool.py", line 313, in aclose
    await self._close_connections(closing_connections)
  File "/home/user/.cache/pypoetry/virtualenvs/app-gEFTwlce-py3.12/lib/python3.12/site-packages/httpcore/_async/connection_pool.py", line 305, in _close_connections
    await connection.aclose()
  File "/home/user/.cache/pypoetry/virtualenvs/app-gEFTwlce-py3.12/lib/python3.12/site-packages/httpcore/_async/connection.py", line 171, in aclose
    await self._connection.aclose()
  File "/home/user/.cache/pypoetry/virtualenvs/app-gEFTwlce-py3.12/lib/python3.12/site-packages/httpcore/_async/http11.py", line 265, in aclose
    await self._network_stream.aclose()
  File "/home/user/.cache/pypoetry/virtualenvs/app-gEFTwlce-py3.12/lib/python3.12/site-packages/httpcore/_backends/anyio.py", line 55, in aclose
    await self._stream.aclose()
  File "/home/user/.cache/pypoetry/virtualenvs/app-gEFTwlce-py3.12/lib/python3.12/site-packages/anyio/streams/tls.py", line 201, in aclose
    await self.transport_stream.aclose()
  File "/home/user/.cache/pypoetry/virtualenvs/app-gEFTwlce-py3.12/lib/python3.12/site-packages/anyio/_backends/_asyncio.py", line 1287, in aclose
    self._transport.close()
  File "/nix/store/gmx7bwrwy6s0kk89ij5yj8r8ayai95x1-python3-3.12.5/lib/python3.12/asyncio/selector_events.py", line 1210, in close
    super().close()
  File "/nix/store/gmx7bwrwy6s0kk89ij5yj8r8ayai95x1-python3-3.12.5/lib/python3.12/asyncio/selector_events.py", line 875, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "/nix/store/gmx7bwrwy6s0kk89ij5yj8r8ayai95x1-python3-3.12.5/lib/python3.12/asyncio/base_events.py", line 795, in call_soon
    self._check_closed()
  File "/nix/store/gmx7bwrwy6s0kk89ij5yj8r8ayai95x1-python3-3.12.5/lib/python3.12/asyncio/base_events.py", line 541, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

问题解答 #

一、错误原因 #

1. 循环不匹配：asyncio.run(main())会创建独立的事件循环，运行完成后自动关闭。而Request单例初始化时，若检测不到运行中的循环，会创建新循环并设为全局循环，但asyncio.run会忽略全局循环，使用自己的循环执行请求。
2. 关闭逻辑冲突：程序结束时，weakref.finalize触发_close()，此时asyncio.run的循环已关闭。但httpx的AsyncClient.aclose()底层依赖anyio，关闭连接时会尝试调用实际运行请求的那个已关闭循环的call_soon方法，从而触发Event loop is closed错误。

二、不使用nest_asyncio的修复方案 #

方案1：调整单例逻辑，绑定实际运行的循环

修改Request类，取消自行创建循环的逻辑，确保Client在实际运行的async上下文初始化，关闭时用正确的循环执行清理：

import asyncio
import weakref
from threading import Lock
from httpx import AsyncClient


class Request:
    _instance = None
    _lock = Lock()
    _session = None
    _loop = None

    def __new__(cls):
        if not cls._instance:
            with cls._lock:
                if not cls._instance:
                    instance = super().__new__(cls)
                    cls._instance = instance
                    weakref.finalize(instance, instance._close)
        return cls._instance

    async def init_client(self):
        if self._session is None:
            self._loop = asyncio.get_running_loop()
            self._session = AsyncClient()

    @classmethod
    def _close(cls):
        if cls._session and cls._loop and not cls._loop.is_closed():
            asyncio.run_coroutine_threadsafe(cls.close(), cls._loop).result()

    @classmethod
    async def close(cls):
        if cls._session:
            await cls._session.aclose()
            cls._session = None

    async def get(self, url):
        if self._session is None:
            await self.init_client()
        try:
            return await self._session.get(url)
        except Exception:
            return None


req = Request()


async def main():
    await req.init_client()
    result = await req.get('https://www.google.com')
    if result:
        print(result.text[:100])


asyncio.run(main())

核心改进：

• 移除自行创建循环的逻辑，让Client绑定到执行请求的循环。
• 关闭时通过run_coroutine_threadsafe确保在正确的循环中执行清理，避免调用已关闭的循环。

方案2：使用异步上下文管理器（推荐）

利用异步上下文管理器规范Client的生命周期，自动处理创建和关闭，彻底避免循环不匹配问题：

import asyncio
from httpx import AsyncClient


class Request:
    _instance = None
    _session = None

    def __new__(cls):
        if not cls._instance:
            cls._instance = super().__new__(cls)
        return cls._instance

    async def __aenter__(self):
        self._session = AsyncClient()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self._session.aclose()

    async def get(self, url):
        try:
            return await self._session.get(url)
        except Exception:
            return None


async def main():
    async with Request() as req:
        result = await req.get('https://www.google.com')
        if result:
            print(result.text[:100])


asyncio.run(main())

优势：

• 上下文管理器自动处理Client的创建和关闭，生命周期清晰。
• 完全避免手动管理循环带来的冲突问题。

三、aiohttp相同逻辑正常运行的原因 #

aiohttp的ClientSession.aclose()实现更健壮：

1. 关闭连接时会检查循环状态，避免调用已关闭的循环。
2. 连接池清理逻辑不依赖第三方库（如httpx依赖的anyio），直接处理资源，不会触发额外的循环调用，因此不会出现该错误。

内容的提问来源于stack exchange，提问作者Babak