celery,flask sqlalchemy: DatabaseError: (DatabaseError) SSL error: 解密失败或坏记录mac

Question

您好，我有一个使用 Celery Flask SqlAlchemy 的设置，但我间歇性地收到此错误：

 (psycopg2.DatabaseError) SSL error: decryption failed or bad record mac

我关注了这个帖子：

Celery + SQLAlchemy : DatabaseError: (DatabaseError) SSL error: decryption failed or bad record mac

还有一些，并添加了 prerun 和 postrun 方法：

@task_postrun.connect
def close_session(*args, **kwargs):
    # Flask SQLAlchemy will automatically create new sessions for you from 
    # a scoped session factory, given that we are maintaining the same app
    # context, this ensures tasks have a fresh session (e.g. session errors 
    # won't propagate across tasks)
    d.session.remove()

@task_prerun.connect
def on_task_init(*args, **kwargs):
    d.engine.dispose()

但我仍然看到此错误。有人解决了吗？

请注意，我在 AWS 上运行它（两台服务器访问同一个数据库）。数据库本身托管在它自己的服务器上（不是 RDS）。我相信运行的芹菜后台任务总数为 6（2+4）。 Flask 前端正在使用 gunicorn 运行。

我的相关主题： https://github.com/celery/celery/issues/3238#issuecomment-225975220

Answer 1

这是我的评论以及其他信息：

我在 AWS 上使用 Celery、SQLAlchemy 和 PostgreSQL，没有这样的问题。 我能想到的唯一区别是我在 RDS 上有数据库。 我认为您可以尝试临时切换到 RDS，以测试是否 问题是否仍然存在。如果它随着 RDS 而消失，那么 您需要查看 PostgreSQL 设置。

根据 RDS 参数，我启用了 SSL：

ssl = 1, Enables SSL connections.
ssl_ca_file = /rdsdbdata/rds-metadata/ca-cert.pem
ssl_cert_file = /rdsdbdata/rds-metadata/server-cert.pem
ssl_ciphers = false, Sets the list of allowed SSL ciphers.
ssl_key_file = /rdsdbdata/rds-metadata/server-key.pem
ssl_renegotiation_limit = 0, integer, (kB) Set the amount of traffic to send and receive before renegotiating the encryption keys.

至于Celery初始化代码，大致是这样的

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

import sqldb

engine = sqldb.get_engine()
cached_data = None

def do_the_work():
    global engine, ruckus_data
    if cached_data is not None:
        return cached_data
    db_session = None
    try:
        db_session = scoped_session(sessionmaker(
            autocommit=False, autoflush=False, bind=engine))
        data = sqldb.get_session().query(
            sqldb.system.MyModel).filter_by(
                my_type = sqldb.system.MyModel.TYPEA).all()
        cached_data = {}
        for row in data:
            ... # put row into cached_data
    finally:
        if db_session is not None:
            db_session.remove()
    return cached_data

然后从 celery 任务中调用这个 do_the_work 函数。 sqldb.get_engine 看起来像这样：

from sqlalchemy import create_engine

_engine = None

def get_engine():
    global _engine
    if _engine:
        return _engine
    _engine = create_engine(config.SQL_DB_URL, echo=config.SQL_DB_ECHO)
    return _engine

最后，config模块中的SQL_DB_URI和SQL_DB_ECHO分别是：

SQL_DB_URL = 'postgresql+psycopg2://%s:%s@%s/%s' % (
    POSTGRES_USER, POSTGRES_PASSWORD, POSTGRES_HOST, POSTGRES_DB_NAME)
SQL_DB_ECHO = False