Postgres、sqlalchemy 和多处理

Question

我对 python 多处理相对较新，并且正在努力解决与此主题相关的许多问题。我最新的问题是多处理、sqlalchemy 和 postgres 的组合。通过这种组合，我有时会得到一个

 sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) SSL error: decryption failed or bad record mac

经过研究，我在文档中发现了这个提示：

https://docs.sqlalchemy.org/en/13/core/pooling.html "这很关键 使用连接池时，以及使用连接池时的扩展 通过 create_engine() 创建的引擎，池连接是 不共享给分叉的进程。 TCP 连接表示为 文件描述符，通常跨进程边界工作， 这意味着这将导致对文件描述符的并发访问 代表两个或多个完全独立的 Python 解释器状态。

有两种方法可以解决这个问题。

第一个是，要么在子进程中创建一个新引擎，要么 在现有引擎上，在孩子之前调用 Engine.dispose() 进程使用任何连接。这将删除所有现有的 池中的连接，以便创建所有新连接。 "

还有这个：

uWSGI, Flask, sqlalchemy, and postgres: SSL error: decryption failed or bad record mac "问题最终是 uwsgi 的分叉。

当使用主进程处理多个进程时，uwsgi 在主进程中初始化应用程序，然后复制 应用到每个工作进程。问题是如果你打开一个 初始化应用程序时的数据库连接，然后你有 多个进程共享同一个连接，这会导致错误 上面。”

我的解释是，当使用多处理时，我必须确保每个进程都使用新引擎。 在我的子进程中，只有一个类可以读写 postgres-db，所以我决定在类中定义一个 slqalchemy 引擎：

class WS_DB_Booker():
    def __init__(self):

        engine_inside_class = create_engine(botpak.bas.dontgitp.bot_engine_string)
        Base_inside_class = declarative_base()
        Base_inside_class.metadata.create_all(engine_inside_class)
        session_factory_inside_class = sessionmaker(bind=engine_inside_class)
        self.DBSession_inside_class = scoped_session(session_factory_inside_class)

    def example_method_to_read_from_db(self):
        try:
            sql_alc_session = self.DBSession_inside_class()
            sql_alc_session.query(and_so_on....

这在第一次试验中运行良好，没有任何问题。但我不确定这是在类中定义引擎的正确方法还是会导致任何问题？

Answer 1

您如何分叉您的进程或哪个组件进行分叉，这真的很难理解。

您需要确保在分叉之后实例化WS_DB_Broker 类！

如果你以错误的方式进行操作（在 fork 之前实例化），那么 Engine 可能已经在它的 Pool 中引用了一些 dbapi 连接。请参阅 SQLAlchemy documentation on working with engines。

为了使您的错误更加明显，您可以执行以下操作：

import os

class WS_DB_Booker():
    def __init__(self):
        # Remember the process id from the time of instantiation. If the
        # interpreter is forked then the output of `os.getpid()` will change.
        self._pid = os.getpid()

        engine_inside_class = create_engine(botpak.bas.dontgitp.bot_engine_string)
        Base_inside_class = declarative_base()
        Base_inside_class.metadata.create_all(engine_inside_class)
        session_factory_inside_class = sessionmaker(bind=engine_inside_class)
        self._session = scoped_session(session_factory_inside_class)

    def get_session():
        if self._pid != os.getpid():
             raise RuntimeError("Forked after instantiating! Please fix!")
        return self._session()

    def example_method_to_read_from_db(self):
        try:
            sql_alc_session = self.get_session()  
            #                      ^^^^^^^^^^^^^
            # this may throw RuntimeError when used incorrectly, thus saving you
            # from your own mistake.
            sql_alc_session.query(and_so_on....