如何在 psycopg2 线程连接类中重新连接到 postgreSQL？ SSL SYSCALL 错误导致的失败：在 Azure 中检测到 EOF？

我们的应用程序运行良好，直到我们将 PostgreSQL 移植到 Azure 中的 Microsoft 数据库。然后，我们的应用程序会定期无故失败，并且到处都会出现 SSL SYSCALL 错误 - 删除等。我们已经尝试了互联网上描述的所有内容 - 使用 keepalive args、RAM、内存和其他所有内容。我们想尝试自动重新建立连接。但我们有一个线程连接池。看过这个帖子Psycopg2 在类内自动重新连接 https://stackoverflow.com/questions/62094660/psycopg2-auto-reconnect-inside-a-class但是我们读取数据库的函数在另一个类中。所以我们有两个问题：

1）SSL SYSCALL错误的原因是什么？我已经搜索了所有线索，并且排除了常见的嫌疑人。 2）如何在线程连接池类中发生故障时重新连接 - >这正在烧瓶应用程序中使用

这是我们的应用程序的结构

class DBClass(object):
    _instance = None
    conn= None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = object.__new__(cls)
            try:
                max_conn = 12
                keepalive_args = { "keepalives": 1, "keepalives_idle": 25, "keepalives_interval": 4,"keepalives_count": 9,
                }
               
                db._instance.pool = psycopg2.pool.ThreadedConnectionPool(3, max_conn, db=,
                                                                               host=, user=,
                                                                               password=,
                                                                               port=, **keepalive_args)

            except Exception as ex:
                db._instance = None
                raise ex

        return cls._instance

    def __enter__(self):
        self.conn= self._instance.pool.getconn()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self._instance.pool.putconn(self.connection)

    def __del__(self):
        self._instance.pool.closeall()

#另一个 python 模块有一个名为 clsEmployee 的类。我们有几十个使用上述数据库类的函数。像这样的东西。

    with DBClass() as db:
        pg_conn = db.connection
        cur = pg_conn.cursor()
        cur.execute("SELECT * from emp")
        row = cur.fetchone()[0]

有很多方法可以解决这个问题。

提出的解决方案Psycopg2 在类内自动重新连接 https://stackoverflow.com/questions/62094660/psycopg2-auto-reconnect-inside-a-class如果您对执行数据库工作的调用位于 DBClass 之外，仍然可以工作。您只需要调用数据库的函数，然后用装饰器包装它们。装饰器所做的就是添加一个循环以允许多次调用该函数，将实际函数包装在 try/ except 中并在 except 上重新连接。这实际上是处理此类问题的一种非常标准的方法，因为它适用于数据库、API 或任何可能失败的东西。您可能想要做的一件事是在重试中添加指数退避（睡眠调用所在的位置）。

您拥有的另一个选项是创建您自己的游标子类，该子类在覆盖的执行版本中具有相同的重试逻辑。这将完成同样的事情，这只是您认为更容易使用的情况。

由于这是在 Flask 应用程序中使用的，因此您可以修改第一种方法，而不是在模型代码级别进行重试，您可以在 Flask 路由级别进行重试。