使用Google Cloud SQL Connector+SQLAlchemy遇Event loop is closed错误求解决

问题场景 #

使用google.cloud.sql.connector连接Google Cloud SQL PostgreSQL实例，查询数据成功，但程序结束时抛出RuntimeError: Event loop is closed错误。

连接代码 #

def getconn():
    with Connector() as connector:
        conn = connector.connect(
            instance_connection_name,
            "pg8000",
            user = db_user,
            password = db_pass,
            db = db_name,
            ip_type = IPTypes.PUBLIC
        )
        return conn

# create connection pool
pool = sqlalchemy.create_engine(
    "postgresql+pg8000://",
    creator=getconn
)

查询代码 #

with pool.connect() as db_conn:
    # query database
    result = db_conn.execute(sqlalchemy.text("SELECT * from users")).fetchall()

    # Do something with the results
    for row in result:
        print(row)
connector.close()

错误信息 #

Exception ignored in: <function _ProactorBasePipeTransport.__del__ at 0x0000026D24CD0A60>
Traceback (most recent call last):
  File "C:\Program Files\Python310\lib\asyncio\proactor_events.py", line 116, in __del__
    self.close()
  File "C:\Program Files\Python310\lib\asyncio\proactor_events.py", line 108, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "C:\Program Files\Python310\lib\asyncio\base_events.py", line 750, in call_soon
    self._check_closed()
  File "C:\Program Files\Python310\lib\asyncio\base_events.py", line 515, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

错误原因 #

1. 无效的connector.close()调用：getconn()中使用with Connector()上下文管理器，退出上下文时会自动关闭Connector实例。外部的connector变量并不存在于当前作用域，调用connector.close()会引用已销毁的对象，触发事件循环残留问题。
2. 连接池资源未正确释放：程序结束时未关闭连接池，导致底层asyncio事件循环的资源未被清理，在Windows环境下的ProactorLoop更容易出现这类问题。

解决方法 #

方法1：移除无效的connector.close()并释放连接池 #

修改查询代码，去掉多余的connector.close()，并在程序结束时调用pool.dispose()释放连接池资源：

with pool.connect() as db_conn:
    result = db_conn.execute(sqlalchemy.text("SELECT * from users")).fetchall()
    for row in result:
        print(row)

# 释放连接池所有资源，关闭活跃连接
pool.dispose()

方法2：（可选）调整Windows异步事件循环策略 #

如果上述方法仍未解决问题（多见于Windows环境），在代码开头添加事件循环策略设置：

import asyncio
# 替换默认的ProactorLoop为SelectorLoop
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

# 后续的Connector和连接池代码...

完整修正代码 #

import asyncio
from google.cloud.sql.connector import Connector, IPTypes
import sqlalchemy

# 适配Windows异步事件循环
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

def getconn():
    with Connector() as connector:
        conn = connector.connect(
            instance_connection_name,
            "pg8000",
            user=db_user,
            password=db_pass,
            db=db_name,
            ip_type=IPTypes.PUBLIC
        )
        return conn

# 创建连接池
pool = sqlalchemy.create_engine(
    "postgresql+pg8000://",
    creator=getconn
)

# 查询数据
with pool.connect() as db_conn:
    result = db_conn.execute(sqlalchemy.text("SELECT * from users")).fetchall()
    for row in result:
        print(row)

# 释放连接池资源
pool.dispose()

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