Celery Worker 数据库连接池

Question

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

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

我使用的是 PostgreSQL，但这同样适用于使用连接的其他存储类型。过去，我使用数据库连接池来避免在每个请求上创建新的数据库连接或避免连接打开时间过长。但是，由于每个 Celery 工作人员都在一个单独的进程中运行，我不确定他们实际上如何能够共享池。我错过了什么吗？我知道 Celery 允许您保留从 Celery 工作人员返回的结果，但这不是我在这里想要做的。根据处理的数据，每个任务可以执行多个不同的更新或插入。

从 Celery worker 中访问数据库的正确方法是什么？

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

Answer 1

我喜欢tigeronk2 的每个worker 一个连接的想法。正如他所说，Celery 维护自己的工作池，因此实际上不需要单独的数据库连接池。 Celery Signal docs 解释了如何在创建 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()

Answer 2

通过实施和监控来回馈我的发现。

欢迎反馈。

参考： 使用池化http://www.prschmid.com/2013/04/using-sqlalchemy-with-celery-tasks.html

每个工作进程（由 -c k 指定的 prefork 模式）将建立一个与 DB 的新连接，而无需池化或重用。 因此，如果使用池化，则只能在每个工作进程级别看到该池。所以池大小 > 1 没有用，但重用连接仍然可以保存打开和关闭的连接。

如果每个工作进程使用一个连接，则在初始化阶段为每个工作进程建立 1 个 DB 连接（prefork 模式 celery -A app worker -c k）。 它可以重复打开和关闭连接。

无论有多少工作线程（eventlet），每个工作线程（celery -A app worker -P eventlet）只建立一个与DB的连接，不进行池化或复用。 因此对于 eventlet，一个 celery 进程（celery -A app worker ...）上的所有工作线程（eventlet）在每个时刻都有 1 个 db 连接。

根据 celery 文档

但您需要确保您的任务不会执行阻塞调用，因为 这将停止工作人员中的所有其他操作，直到阻塞 调用返回。

可能是由于MYSQL DB连接方式阻塞调用。

Answer 3

也许你可以使用pgbouncer。对于 celery，什么都不应该改变，并且连接池是在进程之外完成的。我有同样的issue。

（'也许'，因为我不确定是否会有任何副作用）

Answer 4

每个工作进程有一个数据库连接。由于 celery 本身维护一个工作进程池，因此您的数据库连接将始终等于 celery 工作人员的数量。 另一方面，它会将数据库连接池绑定到 celery 工作进程管理。但这应该没问题，因为 GIL 在一个进程中一次只允许一个线程。

Answer 5

也许，celery.concurrency.gevent 可以提供池共享并且不会加剧 GIL。但是，它的支持仍然是“实验性的”。

还有一个psycopg2.pool.SimpleConnectionPool 在greenlets（协程）之间共享，它们都将在单个进程/线程中运行。

关于该主题的其他stack 讨论。

Answer 6

您可以覆盖默认行为，在 celery 配置中让线程工作人员而不是每个进程的工作人员：

CELERYD_POOL = "celery.concurrency.threads.TaskPool"

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