如何在Celery中实现进程级变量优化数据库连接池？

你当前的模块级db_connection_pool会被所有Celery Worker进程共享（准确说是fork后每个进程会有初始副本，但后续修改各自独立，这种方式不够可靠），要实现进程级独立的连接池缓存，可以用以下几种方案：

方案1：使用multiprocessing.local()（推荐） #

multiprocessing.local()是Python标准库提供的进程级局部存储工具，每个进程访问该对象时会获取自己独有的实例，天然实现进程隔离：

from multiprocessing import local

# 创建进程级局部存储容器
local_store = local()

def get_connection(some_id):
    # 初始化当前进程的连接池（仅第一次调用时执行）
    if not hasattr(local_store, 'db_connection_pool'):
        local_store.db_connection_pool = {}
    
    conn_pool = local_store.db_connection_pool
    if conn := conn_pool.get(some_id):
        return conn
    # 创建新连接并缓存到当前进程的连接池
    conn, some_id = create_connection()
    conn_pool[some_id] = conn
    return conn

每个Celery Worker进程都会维护自己的db_connection_pool，完全不会和其他进程产生干扰。

方案2：利用Celery进程初始化信号 #

通过Celery的worker_process_init信号，在每个Worker进程启动时初始化专属的连接池：

from celery import Celery
from celery.signals import worker_process_init

app = Celery('your_task_app')

# 进程级连接池变量，每个进程启动后会被重新初始化
db_connection_pool = None

@worker_process_init.connect
def init_worker_conn_pool(**kwargs):
    global db_connection_pool
    # 每个Worker进程启动时，创建自己的空连接池
    db_connection_pool = {}

def get_connection(some_id):
    if conn := db_connection_pool.get(some_id):
        return conn
    conn, some_id = create_connection()
    db_connection_pool[some_id] = conn
    return conn

该信号会在每个Worker进程启动时触发，确保db_connection_pool是当前进程独有的变量。

方案3：使用contextvars（Python 3.7+） #

contextvars不仅支持线程级隔离，在多进程环境下也会自动实现进程隔离，适合同时需要线程和进程隔离的场景：

import contextvars

# 创建上下文变量，默认值为空连接池
db_conn_pool_var = contextvars.ContextVar('db_connection_pool', default={})

def get_connection(some_id):
    conn_pool = db_conn_pool_var.get()
    if conn := conn_pool.get(some_id):
        return conn
    # 创建新连接后，更新当前上下文的连接池
    conn, some_id = create_connection()
    new_pool = conn_pool.copy()
    new_pool[some_id] = conn
    db_conn_pool_var.set(new_pool)
    return conn

注意事项 #

• 数据库连接不能跨进程共享，必须保证每个进程使用自己的连接池，否则会出现连接失效、数据错乱等问题；
• 可以配合worker_process_shutdown信号，在Worker进程退出时关闭所有连接，避免资源泄漏；
• 如果使用Celery的gevent/eventlet并发模式（协程），需要改用协程级的隔离方案（比如gevent.local），以上方案仅针对prefork多进程模式。

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