python - 具有异步任务和redis的django中的线程安全

Question

我有一个 Django 应用程序调用查询集上的异步任务(使用 celery )。该任务获取查询集并执行一大堆操作，这些操作可能会根据其中的对象花费很长时间。对象可以跨查询集共享，因此用户可以在包含已经运行的对象的查询集上提交任务，并且该新任务应该只在尚未运行的对象上执行，但等待所有对象完成在它返回之前。

我的解释有点困惑，想象一下下面的代码:

from time import sleep
import redis
from celery.task import Task
from someapp.models import InterestingModel
from someapp.longtime import i_take_a_while

class LongRunningTask(Task):
    def run(self, process_id, *args, **kwargs):
        _queryset = InterestingModel.objects.filter(process__id=process_id)

        r = redis.Redis()
        p = r.pipeline()
        run_check_sets = ('run_check', 'objects_already_running')

        # There must be a better way to do this:
        for o in _queryset.values_list('pk', flat=True):
            p.sadd('run_check')
        p.sdiff(run_check_sets) # Objects that need to be run
        p.sunion(run_check_sets) # Objects that we need to wait for
        p.sunionstore('objects_already_running',run_check_sets)
        p.delete('run_check')
        redis_result = p.execute()

        objects_to_run = redis_result[-3]
        objects_to_wait_for = redis_result[-2]

        if objects_to_run:
            i_take_a_while(objects_to_run)
            p = r.pipeline()
            for o in objects_to_run:
                p.srem('objects_already_running', o)
            p.execute()

        while objects_to_wait_for:
            p = r.pipeline()
            for o in objects_to_wait_for:
                p.sismember('objects_already_running',o)
            redis_result = p.execute()
            objects_to_wait_for = [objects_to_wait_for[i] for i, member in enumerate(redis_result) if member]
            # Probably need to add some sort of timeout here or in redis
            sleep(30) 

我对 Redis 非常陌生，所以我的主要问题是是否有更有效的方法来操作 Redis 以达到相同的结果。更广泛地说，我想知道 Redis 是否是处理此问题的必要/正确方法。似乎应该有更好的方法来将 Django 模型与 Redis 进行交互。最后，我想知道这段代码实际上是否是线程安全的。任何人都可以在我的逻辑中打出任何漏洞吗？

欢迎任何评论。

Answer 1

您是否可以稍微不同地构建它？具体来说，我会启动每个对象的任务，然后将有关长时间运行的作业的信息存储在某个地方(例如，数据库、缓存等)。当每个单独的对象完成时，它会更新长时间运行的作业信息并检查是否所有作业都已返回。如果是这样，那么当长时间运行的任务完成时，您可以运行任何需要运行的代码。

这样做的好处是在等待其他事情发生时不会占用服务器上的线程。在客户端，您可以定期检查长时间运行的作业的状态，如果需要，甚至可以使用完成的对象数来更新进度表。