django/celery：在 150k Django 对象上运行任务的最佳实践？答案

【问题标题】：django/celery: Best practices to run tasks on 150k Django objects?django/celery：在 150k Django 对象上运行任务的最佳实践？
【发布时间】：2011-11-21 12:57:48
【问题描述】：

我必须在大约 150k Django 对象上运行任务。做这个的最好方式是什么？我使用 Django ORM 作为代理。数据库后端是 MySQL，在所有任务的 task.delay() 期间阻塞和死亡。相关的，我也想从提交表单开始，但结果请求产生了一个非常长的响应时间，超时。

【问题讨论】：

这正是我需要的问题。非常感谢。

标签： django orm celery

【解决方案1】：

我使用 beanstalkd (http://kr.github.com/beanstalkd/) 作为引擎。如果你使用 django-beanstalkd，对于 Django 来说，添加一个 worker 和一个任务非常简单：https://github.com/jonasvp/django-beanstalkd/

对我的使用来说非常可靠。

工人示例：

import os
import time

from django_beanstalkd import beanstalk_job


@beanstalk_job
def background_counting(arg):
    """
    Do some incredibly useful counting to the value of arg
    """
    value = int(arg)
    pid = os.getpid()
    print "[%s] Counting from 1 to %d." % (pid, value)
    for i in range(1, value+1):
        print '[%s] %d' % (pid, i)
        time.sleep(1)

启动工作/工作者/任务：

from django_beanstalkd import BeanstalkClient
client = BeanstalkClient()

client.call('beanstalk_example.background_counting', '5')

（源码摘自django-beanstalkd的示例应用）

享受吧！

【讨论】：

【解决方案2】：

除了将数据库用作“代理”之外，我还会考虑使用其他方式。实在不适合这种工作。

不过，您可以通过启动一个任务来创建其他任务，从而将部分开销从请求/响应周期中移出：

from celery.task import TaskSet, task

from myapp.models import MyModel

@task
def process_object(pk):
    obj = MyModel.objects.get(pk)
    # do something with obj

@task
def process_lots_of_items(ids_to_process):
    return TaskSet(process_object.subtask((id, ))
                       for id in ids_to_process).apply_async()

另外，因为您可能没有 15000 个处理器来处理所有这些对象同时，您可以将对象分成 100 个或 1000 个的块：

from itertools import islice
from celery.task import TaskSet, task
from myapp.models import MyModel

def chunks(it, n):
    for first in it:
        yield [first] + list(islice(it, n - 1))

@task
def process_chunk(pks):
    objs = MyModel.objects.filter(pk__in=pks)
    for obj in objs:
        # do something with obj

@task
def process_lots_of_items(ids_to_process):
    return TaskSet(process_chunk.subtask((chunk, ))
                       for chunk in chunks(iter(ids_to_process),
                                           1000)).apply_async()

【讨论】：

谢谢提问！有了@AppPeL 和您的建议，我转向了 RabbitMQ 代理并立即看到了收获。此外，结合这些任务确定了我想要实现的目标！
你能评论一下这段代码是如何工作的以及它产生了多少任务吗？在过去的 4 天里，我已经尝试过它，但它一直在降低我的数据库。我只在运行时使用了 80% 的 CPU，但在运行一段时间后出现“MySQL 已消失”并连接错误。
听起来您的数据库正在超载？您是否将结果存储在数据库中？这是 django-celery 的默认设置。如果您不需要它们，您应该设置@task(ignore_result=True) 或使用CELERY_IGNORE_RESULT=True 全局禁用它们

【解决方案3】：

尝试使用 RabbitMQ。

很多大公司都在使用 RabbitMQ，人们非常依赖它，因为它是一个非常棒的代理。

Here is a great tutorial on how to get you started with it.

【讨论】：