django + celery的队列，路由与弹性

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

#celery_app.py

#!/usr/bin/env python

# -*- coding: utf-8 -*-

from __future__ import absolute_import

import os

from celery import Celery, platforms

from django.conf import settings

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'project.settings')

app = Celery('task')

app.config_from_object('config')

app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

platforms.C_FORCE_ROOT = True

# config.py

#!/usr/bin/env python

# -*- coding: utf-8 -*-

BROKER_URL = 'redis://***@127.0.0.1:6379/0'

BROKER_TRANSPORT = "redis"

CELERY_RESULT_BACKEND = BROKER_URL

CELERY_QUEUES = {

    "task_1": {

        "exchange": "task_1",

        "exchange_type": "direct",

        "routing_key": "task_1"

    },

    "task_2": {

        "routing_key": "task_2",

        "exchange_type": "direct",

        "exchange": "task_2",

    }

}

CELERY_ROUTES = {

    'tasks.task_1': {'queue': 'task_1', 'routing_key': 'task_1'},

    'tasks.task_2': {'queue': 'task_2', 'routing_key': 'task_2'}

}

# run_celery.sh

python manage.py celery worker -A celery_app -P gevent -c  -n task_1_worker -Ofair -l error -Q task_1 -f /home/test/logs/celery.log 

# 同一个manager.py不能同时起gevent和prefork模式，因为monkey_patch，需要根据入参动态适配

python manage.py celery worker -A celery_app -c  --autoscale=, -n task_2_worker  -Ofair -l error -Q task_2 -f /home/test/logs/celery.log