为每个芹菜工人创建单独的数据库连接

Question

当工作人员在创建之后执行任务时，我不断遇到wierdmysql问题。

我们使用django 1.3，芹菜3.1.17，djorm ext pool 0.5

我们用并发3启动芹菜进程。到目前为止，我的观察是，当工人流程开始时，他们都得到相同的mysql定义。我们将html" target="_blank">数据库连接id记录如下。

from django.db import connection
connection.cursor()
logger.info("Task %s processing with db connection %s", str(task_id), str(connection.connection.thread_id()))

当所有工作人员都获得任务时，第一个任务成功执行，但其他两个任务会出现奇怪的Mysql错误。它要么是“Mysql服务器消失”的错误，要么是Django抛出“DoesNotExist”错误的情况。显然，Django查询的对象确实存在。

在这个错误之后，每个worker开始获得它自己的数据库连接，之后我们没有发现任何问题。

芹菜的默认行为是什么？是否设计为共享相同的数据库连接。如果是，如何处理进程间通信？理想情况下，我希望每个工作人员有不同的数据库连接。

我尝试了下面链接中提到的代码，但没有成功。芹菜工人数据库连接池

我们还修复了下面建议的芹菜代码。https://github.com/celery/celery/issues/2453

对于那些投反对票的人，请告诉我投反对票的原因。

Answer 1

芹菜是用下面的命令启动的

celery -A myproject worker --loglevel=debug --concurrency=3 -Q testqueue

myproject.py作为主进程的一部分，在分支工作进程之前对mysql数据库进行一些查询。

作为主进程中查询流的一部分，django ORM创建一个sqlalchemy连接池（如果它还不存在的话）。然后创建工作进程。

芹菜作为django fixups的一部分关闭现有连接。

    def close_database(self, **kwargs):
    if self._close_old_connections:
        return self._close_old_connections()  # Django 1.6
    if not self.db_reuse_max:
        return self._close_database()
    if self._db_recycles >= self.db_reuse_max * 2:
        self._db_recycles = 0
        self._close_database()
    self._db_recycles += 1

实际上，可能发生的情况是，具有一个未使用的db连接的sqlalChemy池对象在分叉时被复制到3个工作进程。因此，3个不同的池有3个指向相同连接文件描述符的连接对象。

Workers在执行任务时，当请求db连接时，所有Worker从sqlalchemy池获得相同的未使用连接，因为该连接当前未使用。所有连接都指向同一个文件描述符，这一事实导致MySQL连接出现错误。

之后创建的新连接都是新的，并且不指向相同的套接字文件描述符。

解决方案：

在主流程中添加

from django.db import connection
connection.cursor()

在完成任何导入之前，即在添加djorm-ext-pool模块之前。

这样，所有的数据库查询都将使用django在池外创建的连接。当芹菜django修复关闭连接时，连接实际上被关闭，而不是回到炼金术池，在分叉时处理所有工人时，炼金术池中没有连接。之后，当工作人员请求db连接时，sqlalChemy将返回一个新创建的连接。