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一起使用芹菜,没有遇到此问题。
这篇关于芹菜/Redis同一任务被多次并行执行的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持跟版网!