SqlAlchemy+Celery带作用域会话

Question

我试图运行一个启动了一堆并行作业的celery_beat作业，但是得到了一个错误：ResourceClosedError: This result object does not return rows. It has been closed automatically.

这是我的相关档案。请注意，我使用的是一个作用域的\u会话：

#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))

^{pr2}$

然后当我尝试初始化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.

我相信我正确地使用了作用域的\u会话。在

有什么建议吗？在

Answer 1

我也犯了同样的错误，还有一些错误，比如：

DatabaseError: server sent data ("D" message) without prior row description ("T" message)
lost synchronization with server: got message type "�", length -1244613424

DatabaseError: lost synchronization with server: got message type "0", length 842674226

原来这是因为我的Celery工作进程共享SQLAlchemy连接。SQLAlchemy docs地址：

It’s critical that when using a connection pool, and by extension when using an Engine created via create_engine(), that the pooled connections are not shared to a forked process. TCP connections are represented as file descriptors, which usually work across process boundaries, meaning this will cause concurrent access to the file descriptor on behalf of two or more entirely independent Python interpreter states.

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

^{pr2}$