Python Celery group() - TypeError: [...] argument after ** must be a mapping, not long

Question

我正在尝试运行一个 celery (3.1.17) 任务，该任务在一个组中执行更多任务，但我总是运行出错。这就是我设置代码的方式：

from celery import task, group

@task
def daily_emails():

    [...]

    all_tasks = []

    for chunk in range(0, users.count(), 1000):
        some_users = users[chunk:chunk+1000]
        all_tasks.append(write_email_bunch.subtask(some_users, execnum))

    job = group(all_tasks)
    # result = job.apply_async()
    # job.get()
    result = job.delay()
    print result
    results = result.join()
    print results

    print "done writing email tasks"
    count = sum(results)
    print count


@task
def write_email_bunch(some_users, execnum):

    [...]

    return len(some_users) - skipped_email_count

这是输出：

<GroupResult: 3d766c85-21af-4ed0-90cb-a1ca2d281db1 [69527252-8468-4358-9328-144f727f372b, 6d03d86e-1b69-4f43-832e-bd27c4dfc092, 1d868d1b-b502-4672-9895-430089e9532e]>
Traceback (most recent call last):
  File "send_daily_emails.py", line 8, in <module>
    daily_emails()
  File "/var/www/virtualenvs/nt_dev/local/lib/python2.7/site-packages/celery/app/task.py", line 420, in __call__
    return self.run(*args, **kwargs)
  File "/var/www/nt_dev/nt/apps/emails/tasks.py", line 124, in daily_emails
    results = result.join()
  File "/var/www/virtualenvs/nt_dev/local/lib/python2.7/site-packages/celery/result.py", line 642, in join
    interval=interval, no_ack=no_ack,
  File "/var/www/virtualenvs/nt_dev/local/lib/python2.7/site-packages/celery/result.py", line 870, in get
    raise self.result
TypeError: write_email_bunch() argument after ** must be a mapping, not long

所以我得到了一个 GroupResult 但不知何故我无法加入它或进一步处理它。当我使用 write_email_bunch.s(some_users, execnum) 我得到这个异常：

  File "/var/www/virtualenvs/nt_dev/local/lib/python2.7/site-packages/celery/result.py", line 870, in get
    raise self.result
TypeError: 'tuple' object is not callable

如何等待所有组任务完成后再继续？ job.get() 给我这个例外：

TypeError: get expected at least 1 arguments, got 0

Answer 1

subtask 接受一个 args 元组，一个 kwargs 字典和任务选项，所以它应该这样调用：

    all_tasks.append(write_email_bunch.subtask((some_users, execnum)))

请注意，我们正在向它传递一个包含 args

的元组

此外，您不应该在任务中等待任务 - 这可能会导致死锁。在这种情况下，我认为 daily_emails 不需要是 celery 任务——它可以是创建 canvas 对象并运行异步应用的常规函数。

def daily_emails():

    all_tasks = []

    for chunk in range(0, users.count(), 1000):
        some_users = users[chunk:chunk+1000]
        all_tasks.append(write_email_bunch.subtask(some_users, execnum))

    job = group(all_tasks)
    result = job.apply_async()
    return result.id

Answer 2

除了其他答案，您还可以在此处使用 chunks： http://docs.celeryproject.org/en/latest/userguide/canvas.html#chunks

@app.task
def daily_emails():
    return write_email.chunks(users, 1000).delay()

@task
def write_email(user):
    [...]

如果同时获取多个对象，手动执行可能会有所帮助来自数据库很重要。您还应该考虑模型对象将在此处序列化，以避免您只能发送 pk 并在任务中重新获取模型，或者发送您关心的字段（例如电子邮件地址或发送该电子邮件所需的任何内容）给用户）。

Python Celery group() - TypeError: [...] argument after ** must be a mapping, not long

Python Celery group() - TypeError: [...] argument after ** must be a mapping, not long

django

celery