芹菜工人数据库连接池

Question

我正在使用Celery独立版(不在Django中).我计划在多台物理机上运行一个工作任务类型.该任务执行以下操作

1. 接受XML文档.
2. 改变它.
3. 进行多个数据库读写操作.

我正在使用PostgreSQL,但这同样适用于使用连接的其他商店类型.过去,我使用数据库连接池来避免在每个请求上创建新的数据库连接,或者避免将连接打开太长时间.但是,由于每个Celery工作程序都在一个单独的进程中运行,我不确定它们实际上是如何共享池的.我错过了什么吗？我知道Celery允许你坚持从芹菜工人那里得到的结果,但这不是我想在这里做的.每个任务可以根据处理的数据执行多个不同的更新或插入.

从Celery工作者访问数据库的正确方法是什么？

是否可以跨多个工作人员/任务共享一个池,还是有其他方法可以做到这一点？

Answer 1

我喜欢tigeronk2关于每个工人一个连接的想法.正如他所说,Celery维护着自己的工作池,因此确实不需要单独的数据库连接池.将芹菜信号文档解释如何做定制的初始化时创建一个工人,所以我下面的代码添加到我的tasks.py,它似乎工作完全像你期望的那样.当工人关机时我甚至能够关闭连接:

db_conn = None

@worker_process_init.connect
def init_worker(**kwargs):
    global db_conn
    print('Initializing database connection for worker.')
    db_conn = db.connect(DB_CONNECT_STRING)


@worker_process_shutdown.connect
def shutdown_worker(**kwargs):
    global db_conn
    if db_conn:
        print('Closing database connectionn for worker.')
        db_conn.close()

Answer 2

您可以覆盖默认行为，在 celery 配置中使用线程工作程序而不是每个进程一个工作程序：

CELERYD_POOL = "celery.concurrency.threads.TaskPool"

然后，您可以将共享池实例存储在任务实例上，并从每个线程任务调用中引用它。