小编Mik*_*ike的帖子

我正在使用Celery和Redis来运行一些后台任务,但每次调用一个任务时,它都会创建一个与Redis的新连接.我在Heroku上,我的Redis to Go计划允许10个连接.我很快就达到了这个限制并且达到了"达到最大客户端数"的错误.

如何确保Celery在单个连接上将任务排队,而不是每次都打开一个新连接？

编辑 - 包括完整的追溯

File "/app/.heroku/venv/lib/python2.7/site-packages/django/core/handlers/base.py", line 111, in get_response
   response = callback(request, *callback_args, **callback_kwargs)

 File "/app/.heroku/venv/lib/python2.7/site-packages/newrelic-1.4.0.137/newrelic/api/object_wrapper.py", line 166, in __call__
   self._nr_instance, args, kwargs)

 File "/app/.heroku/venv/lib/python2.7/site-packages/newrelic-1.4.0.137/newrelic/hooks/framework_django.py", line 447, in wrapper
   return wrapped(*args, **kwargs)

 File "/app/.heroku/venv/lib/python2.7/site-packages/django/views/decorators/csrf.py", line 77, in wrapped_view
   return view_func(*args, **kwargs)

 File "/app/feedback/views.py", line 264, in zencoder_webhook_handler
   tasks.process_zencoder_notification.delay(webhook)

 File "/app/.heroku/venv/lib/python2.7/site-packages/celery/app/task.py", line 343, in delay
   return self.apply_async(args, kwargs)

 File "/app/.heroku/venv/lib/python2.7/site-packages/celery/app/task.py", line 458, in apply_async
   with app.producer_or_acquire(producer) as P:

 File "/usr/local/lib/python2.7/contextlib.py", line 17, in __enter__
   return self.gen.next()

 File …

Celery的连接会定期重置,并显示以下错误消息:

[Errno 104] Connection reset by peer

目前,我失去了任务以及之后触发的任何事情,直到我重置服务器.我有两个问题:(1)导致错误的原因/如何防止错误？(2)如何在错误发生时捕获错误并重试我的任务？

我在Heroku上运行.Celery配置了AMQP(CloudAMQP)作为代理,redis(REDIS_TO_GO)作为结果后端.

堆栈跟踪:

File "celery/task/trace.py", line 224, in trace_task
  R = retval = fun(*args, **kwargs)
File "myapp/tasks.py", line 310, in mytask1
  mytask2.delay(message)
File "celery/app/task.py", line 343, in delay
  return self.apply_async(args, kwargs)
File "celery/app/task.py", line 458, in apply_async
  with app.producer_or_acquire(producer) as P:
File "python2.7/contextlib.py", line 17, in __enter__
  return self.gen.next()
File "celery/app/base.py", line 256, in producer_or_acquire
  with self.amqp.producer_pool.acquire(block=True) as producer:
File "kombu/connection.py", line 712, in acquire
  R = self.prepare(R)
File "kombu/pools.py", line 57, in …