芹菜工作者数据库连接池

Question

我使用芹菜独立（不在Django内）。我计划在多台物理机器上运行一种辅助任务类型。该任务执行以下操作

1. 接受XML文档。
2. 转换它。
3. 进行多个数据库读写

我正在使用PostgreSQL，但这同样适用于使用连接的其他存储类型。过去，我使用数据库连接池来避免在每次请求时创建新的数据库连接，或者避免连接打开时间过长。然而，由于每个芹菜工人都在一个单独的过程中运行，我不确定他们实际上如何能够共享游泳池。我是不是漏了什么？我知道芹菜允许你坚持从芹菜工人返回的结果，但这不是我在这里试图做的。根据处理的数据，每个任务可以执行几个不同的更新或插入。

从芹菜工人内部访问数据库的正确方法是什么？

是否可以跨多个员工/任务共享一个池，或者是否有其他方法来实现这一点？

Answer 1

您可以在芹菜配置中覆盖默认行为，使每个进程具有线程工作线程，而不是工作线程：

CELERYD\u POOL=“celery.concurrency.threads.TaskPool”

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

Answer 2

每个辅助进程有一个DB连接。由于芹菜本身维护一个辅助进程池，因此您的数据库连接将始终等于芹菜辅助进程的数量。从反面来说，它将把数据库连池与芹菜工人的过程管理联系起来。但是考虑到GIL在一个进程中一次只允许一个线程，这应该没问题。

Answer 3

我喜欢tigeronk2关于每个工人一个连接的想法。正如他所说，芹菜维护自己的工作人员池，因此实际上不需要单独的数据库连接池。芹菜信号文档解释了如何在创建worker时进行自定义初始化，因此我将以下代码添加到了我的tasks.py中，它的工作方式似乎与您预期的完全相同。我甚至可以在工人停工时关闭连接：

from celery.signals import worker_process_init, worker_process_shutdown

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()