Celery Worker 数据库连接池

Posted 2023-02-16

技术标签:

【中文标题】Celery Worker 数据库连接池【英文标题】：Celery Worker Database Connection Pooling 【发布时间】：2013-01-09 16:17:58 【问题描述】：

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

多次

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

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

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

【问题讨论】：

你解决了吗？我会对解决方案感兴趣。我为每个工作人员设置了一个数据库连接。 @oneself 如果你接受答案就好了嗨，您是如何为每个工作人员获得一个数据库连接的。我会对解决方案感兴趣 【参考方案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()

【讨论】：

我对这个有点困惑。这不会在模块级别创建一个连接，所以对于每个celery worker？如果您从任务中访问全局 db_conn，您将在多个进程或线程中重用相同的连接（取决于您的并发设置。）。它将启动许多连接，但只是覆盖与启动的每个新工作人员的连接。我在这里错过了什么吗？是的，这就是重点：每个工作人员一个连接。这是一个工作进程池的解决方案，在这种情况下，全局 db_conn 仅在进程上下文中是全局的，因此一切都很好。已经有一段时间了，所以我什至不记得我是否费心用线程来测试它。 @ThatAintWorking 这对我来说很好用 pyodbc。它的缺点是，如果你有固定数量的工人，空闲工人的数据库连接可能会变得陈旧和无效。【参考方案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 文档