芹菜/Redis同一任务被多次并行执行

问题描述

我有两个自定义任务(TaskA和TaskB)，它们都继承自celery.Task。调度程序时不时地启动TaskA，而TaskA每次启动N次TaskB使用不同的参数。但由于某些原因，有时使用相同参数的相同TaskB会同时执行两次，这会导致数据库出现不同的问题。

class TaskA(celery.Task):

    def run(self, *args, **kwargs):
        objects = MyModel.objects.filter(processed=False)
                                 .values_list('id', flat=True)
        task_b = TaskB()
        for o in objects:
            o.apply_async(args=[o, ])

class TaskB(celery.Task):

    def run(self, obj_id, *args, **kwargs):
        obj = MyModel.objects.get(id=obj_id)
        # do some stuff with obj

我尝试过的内容

我尝试使用celery.group，希望它能解决此类问题，但我得到的都是错误，说run接受2个参数，但没有提供任何参数。

这是我尝试使用celery.group启动TaskB的方式：

# somewhere in TaskA
task_b = TaskB()
g = celery.group([task_b.s(id) for id in objects])
g.apply_async()

我也是这样试过的：

# somewhere in TaskA
task_b = TaskB()
g = celery.group([task_b.run(id) for id in objects])
g.apply_async()

在g.apply_async()之前执行任务的位置。

问题

问题是来自我启动任务的方式，还是其他原因？这是正常行为吗？

其他信息

在我的本地计算机上，我运行celery 3.1.13和RabbitMQ 3.3.4，在服务器上运行celery 3.1.13和Redis 2.8.9。 在本地机器上我看不到这样的行为，每个任务都执行一次。在服务器上，我看到1-10个这样的任务连续执行两次。

我在本地计算机和服务器上运行芹菜的方式如下：

celery_beat: celery -A proj beat -l info

celery1: celery -A proj worker -Q default -l info --purge -n default_worker -P eventlet -c 50

celery2: celery -A proj worker -Q long -l info --purge -n long_worker -P eventlet -c 200

可行的解决方法

我根据它收到的参数在TaskB上引入了一个锁。经过大约10个小时的测试，我看到到底执行了两次什么，但是锁防止了数据库上的冲突。 这确实解决了我的问题，但我仍然想了解为什么会发生这种情况。

---

解决方案

芹菜Using Redis文档中介绍的fanout_prefix和fanout_patterns设置了吗？我正在和Redis一起使用芹菜，没有遇到此问题。