SqlAlchemy + Celery 出现范围会话错误

Question

我正在尝试运行一个 celery_beat 作业来启动一堆并行作业，但收到错误：ResourceClosedError: This result object does not return rows. It has been closed automatically.

这是我的相关文件。请注意，我使用的是scoped_session：

#db.py
engine = create_engine(SETTINGS['DATABASE_URL'], pool_recycle=3600, pool_size=10)
db_session = scoped_session(sessionmaker(
    autocommit=False, autoflush=False, bind=engine))

#tasks.py
from db import db_session
@app.task
def db_task(pid):

    db_session()
    r = db_session.query(exists().where(RSSSummary.id == pid)).scalar()

    print pid, r
    db_session.remove()


@app.task
def sched_test():
    ids =[0, 1]

    db_task.delay(ids[0])
    db_task.delay(ids[1])

然后当我尝试启动时sched_test，就像这样：

>>> tasks.sched_test.delay()

DatabaseError: (psycopg2.DatabaseError) error with status PGRES_TUPLES_OK and no message from the libpq

和

ResourceClosedError: This result object does not return rows. It has been closed automatically.

我相信我正确使用了scoped_sessions。

有什么建议么？

Answer 1

我遇到了同样的错误以及一些错误，例如：

\n\n

DatabaseError: server sent data ("D" message) without prior row description ("T" message)\nlost synchronization with server: got message type "\xef\xbf\xbd", length -1244613424\n\nDatabaseError: lost synchronization with server: got message type "0", length 842674226\n

\n\n

事实证明，这是因为我的 Celery 工作进程正在共享 SQLAlchemy 连接。 SQLAlchemy 文档解决了这个问题：

\n\n

\n

至关重要的是，当使用连接池时，以及通过扩展使用通过 create_engine() 创建的引擎时，池中的连接不会共享给分叉进程。TCP 连接表示为文件描述符，通常跨进程边界工作，这意味着这将导致代表两个或多个完全独立的 Python 解释器状态对文件描述符进行并发访问。

\n

\n\n

我通过使用 Celery 事件在工作进程启动时使池中的所有现有连接失效来修复此问题：

\n\n

DatabaseError: server sent data ("D" message) without prior row description ("T" message)\nlost synchronization with server: got message type "\xef\xbf\xbd", length -1244613424\n\nDatabaseError: lost synchronization with server: got message type "0", length 842674226\n

\n