在 Django 中向长时间运行的方法发送信号答案

【问题标题】：Sending signal to long running method in Django在 Django 中向长时间运行的方法发送信号
【发布时间】：2013-04-12 22:50:24
【问题描述】：

我想向 Celery 中长时间运行的任务发送“暂停”信号，我正在尝试找出最好的方法。在视图中，我可以从数据库中提取对象的一个实例并告诉它保存，但它与 Celery 中的对象实例不同。该对象不会检查它是否已暂停。

从长期运行的类和任务中轮询数据库感觉很奇怪且不切实际，因此我正在考虑向我的实例发送消息。我看过使用 pubsub，但我更喜欢使用 Django 信号，因为它已经是一个 Django 项目。我可能以错误的方式处理这个问题。

这是一个不起作用的例子：

模型.py

class LongRunningClass(models.Model):
    is_paused = models.BooleanField(default=False)
    processed_files = models.IntegerField(default=0)
    total_files = models.IntegerField(default=100)

    def long_task(self):
        remaining_files = self.total_files - self.processed_files
        for i in xrange(remaining_files):
            if not self.is_paused:
                self.processed_files += 1
                time.sleep(1)

        # Task complete, let's save.
        self.save()

Views.py

def pause_task(self, pk):
     lrc = LongRunningClass.objects.get(pk=pk)
     lrc.is_paused = True
     lrc.save()
     return HttpResponse(json.dumps({'is_paused': lrc.is_paused}))


def resume_task(self, pk):
    lrc = LongRunningClass.objects.get(pk=pk)
    lrc.is_paused = False
    lrc.save()

    # Pretend this is a Celery task
    lrc.long_task()

因此，如果我修改 models.py 以使用信号，我可以添加这些行，但它仍然无法正常工作：

pause_signal = django.dispatch.Signal(providing_args=['is_paused'])

@django.dispatch.receiver(pause_signal)
def pause_callback(sender, **kwargs):
    if 'is_paused' in kwargs:
        sender.is_paused = kwargs['is_paused']
        sender.save()

这也不影响已经在运行的实例化类。如何让在任务中运行的模型实例暂停？

【问题讨论】：

我从未使用过 celery，但我注意到在您的示例中设置 is_paused 只会导致任务错误地完成（它会循环而不做任何工作）。要暂停执行，它必须是一个阻塞调用......
我更新了我的示例，以更好地说明“暂停”的工作原理。

标签： python django signals celery

【解决方案1】：

Celery 任务是一个单独的进程。 Django 信号是标准的“观察者”模式，它在一个线程中工作，因此无法使用信号来组织线程之间的通信。您需要从数据库中加载对象以了解其属性是否已更改。

class LongRunningClass(models.Model):
    is_paused = models.BooleanField(default=False)
    processed_files = models.IntegerField(default=0)
    total_files = models.IntegerField(default=100)

    def get_is_paused(self):
        db_obj = LongRunningClass.objects.get(pk=self.pk)
        return db_obj.is_paused

    def long_task(self):
        remaining_files = self.total_files - self.processed_files
        for i in xrange(remaining_files):
            if not self.get_is_paused:
                self.processed_files += 1
                time.sleep(1)

    # Task complete, let's save.
    self.save()

设计不是很好 - 你最好将long_task 移动到其他地方，并使用新加载的LongRunningClass 实例进行操作，但它会完成这项工作。您可以在此处添加一些内存缓存 - 如果您不想经常打扰您的数据库。

顺便说一句：我不能 100% 确定，但您可能在这里遇到了另一个设计问题。当您在这种循环中运行非常长时间的任务时，这种情况相当罕见。考虑从你的程序中删除循环（你有队列！）。看看：

@celery.task(run_every=2minutes)  # adding XX files for processing every XX minutes
def scheduled_task(lr_pk):
    lr = LongRunningClass.objects.get(pk=lr_pk)
    if not lr.is paused:
        remaining_files = self.total_files - self.processed_files
        for i in xrange(lr.files_per_iteration):
            process_file.delay(lr.pk,i)

@celery.task(rate=1/m,queue='process_file')  # processing each file
def process_file(lr_pk,i):
    #  do somthing with i
    lr = LongRunningClass.objects.get(pk=lr_pk)
    lr.processed_files += 1
    lr.save()

您必须设置 celerybeat，并为此类任务创建单独的队列，才能实现此解决方案。但结果是，您将对您的程序有很大的控制权——速度、并行执行以及您的代码不会因sleep(1) 而挂起。如果您为每个文件创建另一个模型，您可以控制哪些文件被处理，哪些不是，处理错误等。

【讨论】：

感谢您的建议，我想我可能会在这种情况下使用 memcached 并考虑您的其他笔记。

【解决方案2】：

看看 celery.contrib.abortable - 这是 Celery 任务的替代基类，它在调用者和任务之间实现信号以处理终止，也可用于实现“暂停”。

当调用者调用abort() 时，会在后端标记一个状态。任务调用self.is_aborted() 以查看是否已设置该特殊状态；然后执行任何适当的操作（终止、暂停、忽略等）。动作在任务的控制之下；这不是自动终止任务。

如果特定任务可以将 ABORT 信号解释为暂停请求，则可以按原样使用。或者您可以扩展该类以提供更多信号，而不仅仅是现有的 ABORT。

【讨论】：