如何在使用线程/多处理时使用简单的 sqlalchemy 调用

Question

问题

我正在编写一个从语料库中读取一组文档的程序（每一行都是一个文档）。每个文档都使用一个函数进行处理processdocument，分配一个唯一的 ID，然后写入数据库。理想情况下，我们希望使用多个进程来完成此操作。逻辑如下：

1. 主例程创建一个新数据库并设置一些表。
2. 主例程设置一组将运行辅助函数的进程/线程。
3. 主程序启动所有进程。
4. 主程序读取语料库，将文档添加到队列中。
5. 每个进程的工作函数循环，从队列中读取文档，使用 从中提取信息processdocument，并将信息写入数据库表中的新条目。
6. 一旦队列为空并且主例程设置了适当的标志（一旦没有更多文档添加到队列中），工作者循环就会中断。

题

我对 sqlalchemy（以及一般的数据库）比较陌生。据我所知，我认为用于在主例程中设置数据库的代码工作正常。我遇到的问题是，我不确定在不与其他进程发生冲突的情况下，为每个进程写入数据库的工作函数到底应该放什么。

没有什么特别复杂的事情：每个进程都从一个受锁保护的 multiprocessing.Value 对象中获得一个唯一的值来分配给一个条目。我只是不确定是否应该将什么传递给工作函数（队列除外），如果有的话。我是否传递了我在主程序中创建的 sqlalchemy.Engine 实例？元数据实例？我是否为每个流程创建一个新引擎？有没有其他一些规范的方法来做到这一点？有什么特别需要我记住的吗？

补充评论

我很清楚我可以不打扰多处理，而是在一个进程中执行此操作，但是我将不得不编写具有多个进程读取数据库的代码，所以我不妨弄清楚该怎么做现在这个。

在此先感谢您的帮助！

Answer 1

MetaData 及其 Table 对象集合应被视为应用程序的固定的、不可变的结构，与您的函数和类定义不同。正如您所知，通过 fork 子进程，您的应用程序的所有模块级结构都跨进程边界存在，而表定义通常属于这一类。

然而，引擎指的是一个 DBAPI 连接池，这些连接通常是 TCP/IP 连接，有时是文件句柄。DBAPI 连接本身通常不能在子进程边界上移植，因此您可能希望为每个子进程创建一个新引擎，或者使用非池化引擎，这意味着您使用的是 NullPool。

您也不应该将 MetaData 与 Engine 进行任何类型的关联，即“绑定”元数据。这种做法虽然在各种过时的教程和博客文章中很突出，但实际上并不是通用的东西，我尽量不强调这种工作方式。

如果您使用的是 ORM，则存在类似的“程序结构/活动工作”二分法，其中您的映射类当然在所有子进程之间共享，但您肯定希望 Session 对象是特定子进程的本地对象 - 这些对应于一个实际的 DBAPI 连接以及许多其他最好保持在操作本地的可变状态。