import djcelery
from kombu import Exchange, Queue
djcelery.setup_loader()
BROKER_URL = 'amqp://guest:guest@192.168.141.129:5672/'
CELERY_RESULT_BACKEND = 'django-db'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_SERIALIZER = 'json'
CELERY_TIME_ZONE = TIME_ZONE
CELERY_ENABLE_UTC = True
# 队列声明
CELERY_QUEUES = {
    "add": {
        "exchange": "a",
        "exchange_type": "direct",
        "routing_key": "1"
    },
    "t": {
        "exchange": "a",
        "exchange_type": "direct",
        "routing_key": "2"
    },
}
# 任务绑定队列
CELERY_ROUTES = {
    'celerytest.tasks.add': {"queue": "add"},
    'celerytest.tasks.timed_task': {"queue": "t"}
}
# 定时任务
from datetime import timedelta
CELERYBEAT_SCHEDULE = {
    "test_timed_task": {
        "task": "celerytest.tasks.timed_task",
        "schedule": timedelta(seconds=5)
    }
}

celery -A projectname worker -l info
# 指定此服务器从add队列取任务
celery -A djangoname worker -l info -n worker1@%h -P solo -Q add

@shared_task
def add(x, y):
    sleep(5)
    return x + y

def test_index(request):
    result = tasks.add.apply_async((2, 3), queue='add', routing_key='1')
    mydict = {'status': '请求成功', 'task_id': result.task_id}
    return render(request, 'test.html', mydict)

def my_on_failure(self,exc,task_id,args,kwargs,einfo):
    print('task failed')
 
CELERY_ANNOTATIONS = {'*':{'on_failure':my_on_failure}}