InterfaceError：连接已关闭（使用 django + celery + Scrapy）

Question

在 Celery 任务中使用 Scrapy 解析功能（有时可能需要 10 分钟）时，我得到了这个。

我使用： - 姜戈==1.6.5 - django-celery==3.1.16 - 芹菜==3.1.16 - psycopg2==2.5.5（我也用过psycopg2==2.5.4）

[2015-07-19 11:27:49,488：CRITICAL/MainProcess] 任务 myapp.parse_items [63fc40eb-c0d6-46f4-a64e-acce8301d29a] 内部错误：InterfaceError（'连接已关闭'，） 回溯（最近一次通话最后）： 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/celery/app/trace.py”，第 284 行，在 trace_task uuid，retval，成功，request=task_request， 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/celery/backends/base.py”，第 248 行，在 store_result 请求=请求，**kwargs） _store_result 中的文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/djcelery/backends/database.py”，第 29 行 追溯=追溯，孩子=self.current_task_children（请求）， 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/djcelery/managers.py”，第 42 行，在 _inner 返回乐趣（*args，**kwargs） 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/djcelery/managers.py”，第 181 行，在 store_result “元”：{“孩子”：孩子}}） 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/djcelery/managers.py”，第 87 行，在 update_or_create 返回 get_queryset(self).update_or_create(**kwargs) 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/djcelery/managers.py”，第 70 行，在 update_or_create obj, created = self.get_or_create(**kwargs) 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/query.py”，第 376 行，在 get_or_create 返回 self.get(**lookup), False 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/query.py”，第 304 行，在 get num = len（克隆） 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/query.py”，第 77 行，在 __len__ self._fetch_all() _fetch_all 中的文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/query.py”，第 857 行 self._result_cache = list(self.iterator()) 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/query.py”，第 220 行，在迭代器中 对于 compiler.results_iter() 中的行： 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/sql/compiler.py”，第 713 行，在 results_iter 对于 self.execute_sql(MULTI) 中的行： 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/models/sql/compiler.py”，第 785 行，在 execute_sql 光标 = self.connection.cursor() 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/backends/__init__.py”，第 160 行，在光标中 光标 = self.make_debug_cursor(self._cursor()) _cursor 中的文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/backends/__init__.py”，第 134 行 返回 self.create_cursor() 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/utils.py”，第 99 行，在 __exit__ 六.reraise（dj_exc_type，dj_exc_value，回溯） _cursor 中的文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/backends/__init__.py”，第 134 行 返回 self.create_cursor() 文件“/home/mo/Work/python/pb-env/local/lib/python2.7/site-packages/django/db/backends/postgresql_psycopg2/base.py”，第 137 行，在 create_cursor 光标 = self.connection.cursor() 接口错误：连接已关闭

Answer 1

伙计们和emanuelcds，

我遇到了同样的问题，现在我更新了我的代码并为 celery 创建了一个新的加载器：

from djcelery.loaders import DjangoLoader
from django import db

class CustomDjangoLoader(DjangoLoader):
    def on_task_init(self, task_id, task):
        """Called before every task."""
        for conn in db.connections.all():
            conn.close_if_unusable_or_obsolete()
        super(CustomDjangoLoader, self).on_task_init(task_id, task)

当然，如果您使用的是 djcelery，那么在设置中也需要这样的设置：

CELERY_LOADER = 'myproject.loaders.CustomDjangoLoader'
os.environ['CELERY_LOADER'] = CELERY_LOADER

我还要测试一下，我会更新的。

Answer 2

不幸的是，这是 django + psycopg2 + celery 组合的问题。 这是一个古老且未解决的问题。

看看这个帖子就明白了： https://github.com/celery/django-celery/issues/121

基本上，当 celery 启动一个 worker 时，它会分叉一个数据库连接 来自 django.db 框架。如果此连接由于某种原因断开，它 不会创建一个新的。芹菜与这个问题无关 一旦无法检测到数据库连接何时断开 使用 django.db 库。当它发生时，Django 不会通知， 因为它只是启动一个连接并收到一个 wsgi 调用（没有 连接池）。我在大型生产中遇到了同样的问题 有很多机器工人的环境，有时，这些 机器失去了与 postgres 服务器的连接。

我解决了将每个 celery 主进程置于 linux 下的问题 supervisord 处理程序和一个观察者并实现了一个装饰器 处理 psycopg2.InterfaceError，当它发生时这个函数 调度系统调用以强制主管正常重启 SIGINT celery 进程。

编辑：

找到了更好的解决方案。我实现了一个这样的芹菜任务基类：

from django.db import connection
import celery

class FaultTolerantTask(celery.Task):
    """ Implements after return hook to close the invalid connection.
    This way, django is forced to serve a new connection for the next
    task.
    """
    abstract = True

    def after_return(self, *args, **kwargs):
        connection.close()

@celery.task(base=FaultTolerantTask)
def my_task():
    # my database dependent code here

我相信它也会解决你的问题。