Celery Worker数据库连接池 [英] Celery Worker Database Connection Pooling

问题描述

我正在独立使用Celery（不在Django中）。我计划在多种物理计算机上运行一种工作程序任务类型。该任务执行以下操作

I am using Celery standalone (not within Django). I am planning to have one worker task type running on multiple physical machines. The task does the following

1. 接受XML文档。


2. 对其进行转换。


3. 进行多个数据库读写。

1. Accept an XML document.
2. Transform it.
3. Make multiple database reads and writes.

我正在使用PostgreSQL，但这同样适用于使用连接的其他商店类型。过去，我使用数据库连接池来避免在每个请求上创建新的数据库连接，或者避免保持连接打开时间过长。但是，由于每位Celery工作人员都在一个单独的进程中运行，因此我不确定他们实际上将如何共享该池。我想念什么吗？我知道Celery允许您保留从Celery工作人员返回的结果，但这不是我要在这里做的。每个任务可以执行不同的更新或插入操作，具体取决于所处理的数据。

I'm using PostgreSQL, but this would apply equally to other store types that use connections. In the past, I've used a database connection pool to avoid creating a new database connection on every request or avoid keeping the connection open too long. However, since each Celery worker runs in a separate process, I'm not sure how they would actually be able to share the pool. Am I missing something? I know that Celery allows you to persist a result returned from a Celery worker, but that is not what I'm trying to do here. Each task can do several different updates or inserts depending on the data processed.

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

What is the right way to access a database from within a Celery worker?

是否可以在多个工作程序/任务之间共享一个池，或者有其他方法可以做到这一点？

Is it possible to share a pool across multiple workers/tasks or is there some other way to do this?

推荐答案

我喜欢tigeronk2每个工人只有一个联系的想法。正如他所说，Celery维护自己的工作人员池，因此确实不需要单独的数据库连接池。 Celery Signal文档解释了如何进行自定义初始化当创建一个工作程序时，我将以下代码添加到我的tasks.py中，它似乎完全按照您的期望工作。当工作人员关闭时，我什至能够关闭连接：

I like tigeronk2's idea of one connection per worker. As he says, Celery maintains its own pool of workers so there really isn't a need for a separate database connection pool. The Celery Signal docs explain how to do custom initialization when a worker is created so I added the following code to my tasks.py and it seems to work exactly like you would expect. I was even able to close the connections when the workers are shutdown:

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

这篇关于Celery Worker数据库连接池的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！