Cloud Function并发调用PostgreSQL异常：仅预运行测试查询后正常

Google Cloud Functions（含本地functions-framework）的全局代码初始化在主线程执行，而API请求由独立工作线程处理。注释掉全局的run_concurrency_test()调用后出现卡死的核心原因：

1. 全局声明的engine和Session（scoped_session）仅完成初始化，从未在主线程中实际使用——scoped_session依赖线程本地存储（TLS），主线程的TLS上下文未被激活。
2. API工作线程首次调用Session()时，scoped_session尝试绑定会话到当前线程，同时cloud-sql-python-connector的连接池需在工作线程中建立底层连接，这个过程触发死锁：连接池初始化等待主线程资源，而主线程已进入空闲状态等待工作线程完成。

此外，全局初始化数据库资源不符合Cloud Functions最佳实践，冷启动时全局代码耗时过长会拖慢启动速度，还可能在实例复用阶段引发线程安全问题。

将数据库相关全局变量设为None，仅在首次API请求时初始化，确保资源在工作线程上下文内正确创建：

修改后的main.py代码：

import functions_framework
import sqlalchemy
import threading
from google.cloud.sql.connector import Connector, IPTypes
from sqlalchemy.orm import sessionmaker, scoped_session

Base = sqlalchemy.orm.declarative_base()

class TestUsers(Base):
    __tablename__ = 'TestUsers'
    
    uuid = sqlalchemy.Column(sqlalchemy.String, primary_key=True)

cloud_sql_connection_name = "myproject-123456:asia-northeast3:tosmedb"

# 全局变量初始化为None，延迟加载
connector = None
engine = None
Session = None

def getconn():
    global connector
    if not connector:
        connector = Connector()
    connection = connector.connect(
        cloud_sql_connection_name,
        "pg8000",
        user="postgres",
        password="redacted",
        db="tosme",
        ip_type=IPTypes.PUBLIC,
    )
    return connection

def init_pool():
    global engine
    if not engine:
        engine_url = sqlalchemy.engine.url.URL.create(
            "postgresql+pg8000",
            username="postgres",
            password="redacted",
            host=cloud_sql_connection_name,
            database="tosme"
        )
        engine = sqlalchemy.create_engine(engine_url, creator=getconn)
        # 创建不存在的表
        Base.metadata.create_all(engine)
    return engine

def get_session():
    global Session
    if not Session:
        engine = init_pool()
        # 准备线程安全的会话工厂
        Session = scoped_session(sessionmaker(bind=engine))
    return Session

def run_concurrency_test():
    def get_user():
      Session = get_session()
      with Session() as session:
          session.query(TestUsers).first()

    print("Simulating concurrent reads...")

    threads = []
    for i in range(2):
        thread = threading.Thread(target=get_user)
        threads.append(thread)
        thread.start()

    # 等待所有线程完成
    for thread in threads:
        thread.join()
        print(f"Thread {thread.name} completed")

    print("Test passed - Threads all completed!\n")

# 注释掉全局测试调用
# run_concurrency_test()

@functions_framework.http
def api(request):
    print("API hit - Calling run_concurrency_test()...")
    run_concurrency_test()
    return "Success"

• 将connector、engine、Session全局变量初始化为None，通过专用函数在首次使用时完成初始化。
• 确保所有数据库资源的初始化在API请求的工作线程上下文内完成，避免主线程与工作线程的资源竞争。
• 保留scoped_session的线程安全特性，同时保证其在正确的线程环境中被激活。

全局代码中调用run_concurrency_test()时，所有数据库资源（engine、Session、连接池）都在主线程中初始化并使用，TLS上下文被正确激活，后续工作线程调用时可直接复用已初始化资源，不会触发死锁。但这种方式不符合Cloud Functions运行模型，生产环境可能导致冷启动超时或资源泄漏。

内容的提问来源于stack exchange，提问作者Dr-Bracket