Django 多处理和数据库连接

Question

背景:

我正在开展一个将 Django 与 Postgres 数据库结合使用的项目。我们也在使用 mod_wsgi 以防万一，因为我的一些网络搜索已经提到了它。在 web 表单提交上，Django View 启动一项需要大量时间的工作(比用户希望等待的时间长)，因此我们通过后台系统调用启动该工作。现在正在运行的作业需要能够读取和写入数据库。因为这项工作需要很长时间，所以我们使用多处理来并行运行它的一部分。

问题:

顶级脚本有一个数据库连接，当它产生子进程时，父进程的连接似乎对子进程可用。然后有一个关于如何在查询之前调用 SET TRANSACTION ISOLATION LEVEL 的异常。研究表明，这是由于试图在多个进程中使用相同的数据库连接。我发现的一个线程建议在子进程开始时调用 connection.close() ，这样 Django 会在需要时自动创建一个新连接，因此每个子进程都会有一个唯一的连接——即不共享。这对我不起作用，因为在子进程中调用 connection.close() 会导致父进程提示连接丢失。

其他发现:

我读到的一些资料似乎表明你不能真正做到这一点，而且多处理、mod_wsgi 和 Django 不能很好地协同工作。我猜这似乎很难相信。

有些人建议使用 celery，这可能是一个长期的解决方案，但我目前无法安装 celery，等待一些审批流程，所以现在不是一个选择。

在 SO 和其他地方找到了几个关于持久数据库连接的引用资料，我认为这是一个不同的问题。

还找到了对 psycopg2.pool 和 pgpool 的引用以及关于 bouncer 的一些内容。诚然，我不明白我在这些书上读到的大部分内容，但它肯定没有让我突然想到我正在寻找的东西。

当前的“解决方法”:

现在，我已经恢复到只连续运行东西，它可以工作，但比我想要的要慢。

关于如何使用多处理并行运行有什么建议吗？似乎如果我可以让 parent 和两个 child 都拥有与数据库的独立连接，一切都会好起来的，但我似乎无法理解这种行为。

谢谢，抱歉太长了!

Answer 1

Multiprocessing复制进程之间的连接对象，因为它fork进程，因此复制父进程的所有文件描述符。也就是说，与 SQL 服务器的连接只是一个文件，您可以在 Linux 中的/proc//fd/... 下看到它。任何打开的文件都将在 fork 进程之间共享。您可以找到更多关于 fork 的信息 here .

我的解决方案只是在启动进程之前简单地关闭数据库连接，每个进程在需要时重新创建连接(在 django 1.4 中测试):

from django import db
db.connections.close_all()
def db_worker():      
    some_paralell_code()
Process(target = db_worker,args = ())

Pgbouncer/pgpool 在多进程的意义上没有连接线程。这是不关闭每个请求连接的解决方案 = 在高负载下加速连接到 postgres。

更新:

要完全消除数据库连接问题，只需将所有与数据库连接的逻辑移动到 db_worker - 我想将 QueryDict 作为参数传递...更好的想法是简单地传递 ID 列表...参见 QueryDict和 values_list('id', flat=True)，并且不要忘记将其转为列表! list(QueryDict) 在传递给 db_worker 之前。多亏了我们不复制模型数据库连接。

def db_worker(models_ids):        
    obj = PartModelWorkerClass(model_ids) # here You do Model.objects.filter(id__in = model_ids)
    obj.run()


model_ids = Model.objects.all().values_list('id', flat=True)
model_ids = list(model_ids) # cast to list
process_count = 5
delta = (len(model_ids) / process_count) + 1

# do all the db stuff here ...

# here you can close db connection
from django import db
db.connections.close_all()

for it in range(0:process_count):
    Process(target = db_worker,args = (model_ids[it*delta:(it+1)*delta]))