如何结合使用eventlet monkey_patch与google.cloud.sql.connector？解决线程冲突报错

问题描述 #

使用Flask+SocketIO组合Google Cloud SQL连接器时，触发以下错误：

google.cloud.sql.connector.exceptions.ConnectorLoopError: Connector event loop is running in current thread!Event loop must be attached to a different thread to prevent blocking code!

相关核心代码：

import eventlet
eventlet.monkey_patch(socket=True)

def getconn():
    project = current_app.config["PROJECT"]
    region = current_app.config["REGION"]
    instance_name = current_app.config["INSTANCE_CONNECTION_NAME"]
    conn = connector.connect(
        f"{project}:{region}:{instance_name}",  # Cloud SQL Instance Connection Name
        "pg8000",
        user=current_app.config.get("DB_USER"),
        password=current_app.config.get("DB_PASS"),
        db=current_app.config.get("DB_NAME"),
        ip_type=IPTypes.PRIVATE,  # IPTypes.PRIVATE for private IP
    )
    return conn

def create_app() -> Flask:
    app = Flask(__name__)
    app.config["SQLALCHEMY_DATABASE_URI"] = "postgresql+pg8000://"
    app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {"creator": getconn}

问题原因 #

Google Cloud SQL Connector内部依赖异步IO事件循环，而eventlet的monkey_patch(socket=True)会替换系统原生的socket和事件循环实现。当在eventlet的协程线程中直接调用connector.connect()时，连接器的事件循环会和eventlet的协程线程抢占资源，导致阻塞冲突，触发上述报错。

解决方法 #

核心思路是将Cloud SQL连接器的连接创建逻辑放到独立线程中执行，避免与eventlet的协程线程共享同一个事件循环。

方案1：使用独立线程执行连接创建

修改getconn函数，通过Python标准库的Thread将连接逻辑放到新线程中运行：

import eventlet
from threading import Thread
from google.cloud.sql.connector import Connector, IPTypes
from flask import current_app, Flask

eventlet.monkey_patch(socket=True)

def getconn():
    conn = None
    connector = Connector()
    
    def _connect_in_thread():
        nonlocal conn
        project = current_app.config["PROJECT"]
        region = current_app.config["REGION"]
        instance_name = current_app.config["INSTANCE_CONNECTION_NAME"]
        conn = connector.connect(
            f"{project}:{region}:{instance_name}",
            "pg8000",
            user=current_app.config.get("DB_USER"),
            password=current_app.config.get("DB_PASS"),
            db=current_app.config.get("DB_NAME"),
            ip_type=IPTypes.PRIVATE,
        )
    
    # 启动线程执行连接逻辑，等待完成后返回连接
    thread = Thread(target=_connect_in_thread)
    thread.start()
    thread.join()
    return conn

def create_app() -> Flask:
    app = Flask(__name__)
    app.config["SQLALCHEMY_DATABASE_URI"] = "postgresql+pg8000://"
    app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {"creator": getconn}
    return app

方案2：使用异步API+独立事件循环

利用连接器的异步APIconnect_async，手动创建一个独立的原生事件循环来执行连接逻辑：

import eventlet
import asyncio
from google.cloud.sql.connector import Connector, IPTypes
from flask import current_app, Flask

eventlet.monkey_patch(socket=True)

def getconn():
    project = current_app.config["PROJECT"]
    region = current_app.config["REGION"]
    instance_name = current_app.config["INSTANCE_CONNECTION_NAME"]
    connector = Connector()
    
    # 创建独立的原生事件循环，避免与eventlet的协程循环冲突
    loop = asyncio.new_event_loop()
    conn = loop.run_until_complete(
        connector.connect_async(
            f"{project}:{region}:{instance_name}",
            "pg8000",
            user=current_app.config.get("DB_USER"),
            password=current_app.config.get("DB_PASS"),
            db=current_app.config.get("DB_NAME"),
            ip_type=IPTypes.PRIVATE,
        )
    )
    loop.close()
    return conn

def create_app() -> Flask:
    app = Flask(__name__)
    app.config["SQLALCHEMY_DATABASE_URI"] = "postgresql+pg8000://"
    app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {"creator": getconn}
    return app

说明 #

两种方案都能有效隔离Cloud SQL连接器的事件循环与eventlet的协程线程，避免阻塞冲突。方案1更直观，适合大多数场景；方案2利用异步API，性能略优，但需要注意事件循环的创建和销毁逻辑。

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