django/celery : Best practices to run tasks on 150k Django objects?

Question

我必须在大约15万个Django对象上运行任务。做这个的最好方式是什么？我正在使用Django ORM作为代理。数据库后端是MySQL，在所有任务的task.delay()期间都会死机。与此相关，我也想从提交表单时开始，但是所产生的请求产生了非常长的响应时间，导致超时。

Answer 1

除了将数据库用作“经纪人”之外，我还考虑使用其他东西。确实不适合此类工作。

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

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()