【问题标题】:Celery: How to use "link_error" if documentation has an error?Celery:如果文档有错误,如何使用“link_error”?
【发布时间】:2015-06-13 20:34:54
【问题描述】:

在 Celery 文档中:http://celery.readthedocs.org/en/latest/userguide/canvas.html#chains 是如何使用 link_error 的示例:

您还可以使用 link_error 参数添加错误回调:

add.apply_async((2, 2), link_error=log_error.s())
add.subtask((2, 2), link_error=log_error.s())

由于只有在使用pickle时才能序列化异常,所以错误 回调将父任务的 id 作为参数:

from __future__ import print_function 
import os 
from proj.celery import app

@app.task 
def log_error(task_id):
    result = app.AsyncResult(task_id)
    result.get(propagate=False)  # make sure result written.
    with open(os.path.join('/var/errors', task_id), 'a') as fh:
        print('--\n\n{0} {1} {2}'.format(
            task_id, result.result, result.traceback), file=fh)

但在此示例中是错误,因为他们在任务内部调用 AsuncResult.get 导致 DEADLOCK 和此日志条目:

/opt/.virtualenvs/spark/lib/python3.4/site-packages/celery/result.py:45: RuntimeWarning: Never call result.get() within a task!
See http://docs.celeryq.org/en/latest/userguide/tasks.html#task-synchronous-subtasks

In Celery 3.2 this will result in an exception being
raised instead of just being a warning.

  warnings.warn(RuntimeWarning(E_WOULDBLOCK))

[2015-06-13 20:30:19,242: WARNING/Worker-4] /opt/.virtualenvs/spark/lib/python3.4/site-packages/celery/result.py:45: RuntimeWarning: Never call result.get() within a task!
See http://docs.celeryq.org/en/latest/userguide/tasks.html#task-synchronous-subtasks

In Celery 3.2 this will result in an exception being
raised instead of just being a warning.

  warnings.warn(RuntimeWarning(E_WOULDBLOCK))

【问题讨论】:

    标签: python django celery


    【解决方案1】:

    请执行此检查表以确保配置正常:

    1. CELERY_IGNORE_RESULT 必须设置为 False
    2. CELERY_BACKEND_RESULT 必须设置,例如'amqp'(用于 RabbitMQ)
    3. @task decorator 不得有选项 ignore_result=True 在每个任务上(但你可以创建 immutable signatures

    最后一个是正确修改的获取结果的任务(allow_join_result):

    from celery.result import allow_join_result
    
    @app.task
    def log_error(task_id):
        with allow_join_result():
            result = app.AsyncResult(task_id)
            result.get(propagate=False)  # make sure result written.
            with open(os.path.join('/var/errors', task_id), 'a') as fh:
                print('--\n\n{0} {1} {2}'.format(
                    task_id, result.result, result.traceback), file=fh)
    

    注意: result.result 在这种情况下 ZeroDivisionError 是字典:

    {
        'exc_type': 'ZeroDivisionError',  # Name of Exception type
        'exc_message': 'division by zero'  # Reason of exception
    }
    

    注意:

    这是 #2652 的问题,当 Celery 忽略错误时在 AsyncResult 上调用 get()

    【讨论】:

      猜你喜欢
      • 2015-11-06
      • 1970-01-01
      • 1970-01-01
      • 2021-04-11
      • 2019-07-16
      • 2015-06-13
      • 2017-08-07
      • 1970-01-01
      • 2021-12-02
      相关资源
      最近更新 更多