如果数据库发生变化，如何停止执行长进程？

Question

我有一个向RabbitMQ 队列发送消息的视图。

message = {'origin': 'Bytes CSV',
           'data': {'csv_key': str(csv_entry.key),
                    'csv_fields': csv_fields
                    'order_by': order_by,
                    'filters': filters}}

...

queue_service.send(message=message, headers={}, exchange_name=EXCHANGE_IN_NAME,
                   routing_key=MESSAGES_ROUTING_KEY.replace('#', 'bytes_counting.create'))

在我的消费者身上，我有一个很长的过程来生成 CSV。

def create(self, data):
    csv_obj = self._get_object(key=data['csv_key'])
    if csv_obj.status == CSVRequestStatus.CANCELED:
        self.logger.info(f'CSV {csv_obj.key} was canceled by the user')
        return

    result = self.generate_result_data(filters=data['filters'], order_by=data['order_by'], csv_obj=csv_obj)
    csv_data = self._generate_csv(result=result, csv_fields=data['csv_fields'], csv_obj=csv_obj)
    file_key = self._post_csv(csv_data=csv_data, csv_obj=csv_obj)

    csv_obj.status = CSVRequestStatus.READY
    csv_obj.status_additional = CSVRequestStatusAdditional.SUCCESS
    csv_obj.file_key = file_key
    csv_obj.ready_at = timezone.now()
    csv_obj.save(update_fields=['status', 'status_additional', 'ready_at', 'file_key'])

    self.logger.info(f'CSV {csv_obj.name} created')

长过程发生在self._generate_csv 内部，因为self.generate_result_data 返回一个queryset，这是惰性的。

如您所见，如果用户在消息开始被消费之前通过端点更改csv_request 的状态，则不会评估进程。我的目标是在执行self._generate_csv 期间让这种情况发生。

到目前为止，我尝试使用Threading，但没有成功。

我怎样才能实现我的目标？

非常感谢！

Answer 1

你为什么不检查 Celery 库？将celery with django 与RabbitMQ backend 一起使用比直接利用rabbitmq 队列要容易得多。

Celery 有一个内置函数 revoke 来终止正在进行的任务：

>>> from celery.task.control import revoke
>>> revoke(task_id, terminate=True)

• related SO answer
• celery docs

对于您的用例，您可能需要类似（代码 sn-ps）：

## celery/tasks.py
from celery import app

@app.task(queue="my_queue")
def create_csv(message):
    # ...snip...
    pass

## main.py
from celery import uuid, current_app

def start_task(task_id, message):
    current_app.send_task(
        "create_csv",
        args=[message],
        task_id=task_id,
    )

def kill_task(task_id):
    current_app.control.revoke(task_id, terminate=True)

## signals.py

from django.dispatch import receiver
from .models import MyModel
from .main import kill_task

# choose appropriate signal to listen for DB change
@receiver(models.signals.post_save, sender=MyModel)
def handler(sender, instance, **kwargs):
    kill_task(instance.task_id)

• 使用celery.uuid生成可以存储在DB或缓存中的任务ID，并使用相同的任务ID来控制任务，即请求终止。

Answer 2

由于 self._generate_csv 是最慢的，显而易见的解决方案是使用此函数。

为此，您可以将 csv 文件的创建分成几个部分。创建每个片段后，检查状态，看看是否可以继续创建文件。最后，将所有部分粘贴到一个完成的文件中。

Here is a method for combining multiple files into one.