主要问题
我正在测试如何处理某些任务失败，例如处理" timelimitexceed"例外，该异常立即杀死了任务并不能"可接收"(是的...我知道" softtimelimit'它不符合我的需求(。
第一种方法
这是我的tasks.py(工人使用--time-limit标志运行(：
import logging
from celery import Celery
import time
app = Celery('tasks', broker='pyamqp://guest@localhost//')
def my_fail(task, exc, req_id, req_args, req_kwargs, einfo, *ext_args, **kwargs):
logger.info("args: %r", req_args)
logger.info("kw: %r", req_kwargs)
@app.task(on_failure=my_fail)
def sum(x, y, delay=0, **kw):
result = x+y
if result == 4:
raise Exception("Some Error")
time.sleep(delay)
return x+y
任务失败时的主要想法，能够根据任务的args/kwargs执行一些处理
例如，如果我运行sum.delay(3, 1, foo="bar")，则升级Exception("Some Error")，并且记录以下内容：
[2019-06-30 17:21:45,120: INFO/Worker-1] args: (3, 1)
[2019-06-30 17:21:45,121: INFO/Worker-1] kw: {'foo': 'bar'}
[2019-06-30 17:21:45,122: ERROR/MainProcess] Task tasks.sum[9e9de032-1469-44e7-8932-4c490fcee2e3] raised unexpected: Exception('Some Error',)
Traceback (most recent call last):
File "/home/apernin/.virtualenvs/dr/local/lib/python2.7/site-packages/celery/app/trace.py", line 240, in trace_task
R = retval = fun(*args, **kwargs)
File "/home/apernin/.virtualenvs/dr/local/lib/python2.7/site-packages/celery/app/trace.py", line 438, in __protected_call__
return self.run(*args, **kwargs)
File "/home/apernin/test/tasks.py", line 89, in sum
raise Exception("Some Error")
Exception: Some Error
请注意，Args/Kwargs由我的on-failure处理程序打印。
现在，如果我运行 sum.delay(3, 2, delay=7) timelimit 是触发的
[2019-06-30 17:23:15,244: INFO/MainProcess] Received task: tasks.sum[8c81398b-4378-401d-a674-a3bd3418ccde]
[2019-06-30 17:23:21,070: ERROR/MainProcess] Task tasks.sum[8c81398b-4378-401d-a674-a3bd3418ccde] raised unexpected: TimeLimitExceeded(5.0,)
Traceback (most recent call last):
File "/home/apernin/.virtualenvs/dr/local/lib/python2.7/site-packages/billiard/pool.py", line 645, in on_hard_timeout
raise TimeLimitExceeded(job._timeout)
TimeLimitExceeded: TimeLimitExceeded(5.0,)
[2019-06-30 17:23:21,071: ERROR/MainProcess] Hard time limit (5.0s) exceeded for tasks.sum[8c81398b-4378-401d-a674-a3bd3418ccde]
[2019-06-30 17:23:21,629: ERROR/MainProcess] Process 'Worker-1' pid:15472 exited with 'signal 15 (SIGTERM)'
请注意，由于on-failure处理程序未拒绝。由于芹菜的困难时间限制的性质，这是可以预期的。
第二种方法
我的第二种方法是使用事件列表。
from celery import Celery
def my_monitor(app):
state = app.events.State()
def announce_failed_tasks(event):
state.event(event)
# task name is sent only with -received event, and state
# will keep track of this for us.
task = state.tasks.get(event['uuid'])
with app.connection() as connection:
recv = app.events.Receiver(connection, handlers={
'task-failed': announce_failed_tasks,
})
recv.capture(limit=None, timeout=None, wakeup=True)
if __name__ == '__main__':
app = Celery(broker='amqp://guest@localhost//')
my_monitor(app)
我唯一能够检索的信息是任务 uuid ，我无法检索任务的名称，args或kwargs(任务对象包含属性，但无都是(。
问题
有没有办法：
- 在困难时限的情况下，将
on_failure处理程序制作？ - 用
task-failed侦听器检索任务的任务/kwargs？
预先感谢
首先，超时是由工人处理的(mainprocess(，并且与任务内发生的失败相同，例如被抛出的例外等。这就是为什么这就是为什么将其视为日志中的MainProcess提出的TimeLimitexceed。因此，不幸的是，您不能依靠相同的逻辑...
但是，您的第二种方法将在追踪正在发生的事情时很有用。
我已经开发了(内部(一个芹菜监测工具，该工具可以抓住所有事件，并与它们一起填充数据库，以便以后我们可以进行各种分析(请参阅平均和最差的运行时间，以获取示例，例如失败等(。
为了从task-failed事件给出的数据中获取所需的详细信息，您还需要记录(将其存储在某些字典中以示例(task-received事件数据。此信息包含您可能需要的ARG，任务名称和所有有用的信息。您可以通过任务UUID将它们相关联。