可以使用任何unittest库同步测试任务。在处理芹菜任务时，我通常会进行2次不同的测试。第一个（如我所建议的，下面是波纹管）是完全同步的，应该确保算法执行应做的工作。第二部分使用整个系统（包括代理），并确保我没有序列化问题或任何其他分发，通信问题。
所以：
from celery import Celerycelery = Celery()@celery.taskdef add(x, y): return x + y
和您的测试：
from nose.tools import eq_def test_add_task(): rst = add.apply(args=(4, 4)).get() eq_(rst, 8)
希望有帮助！
