python - 如何处理 ProcessPool 中的 SQLAlchemy 连接？

Question

我有一个 react 器，它从 RabbitMQ 代理获取消息并触发工作方法以在进程池中处理这些消息，如下所示:

这是使用 python asyncio、loop.run_in_executor() 和 concurrent.futures.ProcessPoolExecutor 实现的。

现在我想使用 SQLAlchemy 在工作方法中访问数据库。大多数情况下，处理将是非常简单和快速的 CRUD 操作。

reactor 开始时每秒会处理 10-50 条消息，因此不能为每个请求都打开一个新的数据库连接。相反，我想为每个进程维护一个持久连接。

我的问题是:我该怎么做？我可以将它们存储在全局变量中吗？ SQA 连接池会为我处理这个问题吗？ react 堆停止时如何清理？

[更新]

• 数据库是带有 InnoDB 的 MySQL。

为什么选择这种带有进程池的模式？

当前实现使用不同的模式，每个消费者在自己的线程中运行。不知何故，这不是很好。已经有大约 200 个消费者在各自的线程中运行，并且系统正在快速增长。为了更好地扩展，想法是分离关注点并在 I/O 循环中使用消息并将处理委托(delegate)给池。当然，整个系统的性能主要受 I/O 限制。但是，在处理大型结果集时，CPU 是一个问题。

另一个原因是“易于使用”。虽然消息的连接处理和消费是异步实现的，但worker中的代码可以是同步的和简单的。

很快就发现，通过工作人员内部的持久网络连接访问远程系统是一个问题。这就是 CommunicationChannels 的用途:在 worker 内部，我可以通过这些 channel 向消息总线授予请求。

我目前的一个想法是以类似的方式处理数据库访问:将语句通过队列传递到事件循环，然后将它们发送到数据库。但是，我不知道如何使用 SQLAlchemy 做到这一点。切入点在哪里？对象通过队列时需要pickled。如何从 SQA 查询中获取这样的对象？与数据库的通信必须异步进行，以免阻塞事件循环。我可以使用例如aiomysql 作为 SQA 的数据库驱动程序？

Answer 1

假设您正在使用orm，在工作进程中。

一个简单的解决方案是使用全局 session跨请求重用:

# db.py
engine = create_engine("connection_uri", pool_size=1, max_overflow=0)
DBSession = scoped_session(sessionmaker(bind=engine)) 

关于 worker 任务:

# task.py
from db import engine, DBSession
def task():
    DBSession.begin() # each task will get its own transaction over the global connection
    ...
    DBSession.query(...)
    ...
    DBSession.close() # cleanup on task end

参数 pool_size 和 max_overflow customize默认 QueuePool由 create_engine 使用。pool_size 将确保您的进程仅在进程池中的每个进程保持 1 个连接。

如果您希望它重新连接，您可以使用 DBSession.remove() 从注册表中删除 session ，并使其在下次使用 DBSession 时重新连接。您还可以使用 Pool 的 recycle 参数使连接在指定的时间后重新连接。

在开发/调试期间，您可以使用 AssertionPool如果从池中 check out 多个连接，则会引发异常，请参阅 switching pool implementations关于如何做到这一点。