celery_app.py

from celery import Celery
from config import CELERY_BACKEND_URL, CELERY_BROKER_URL, CELERY_DEFAULT_QUEUE
from kombu import Exchange, Queue
from celery_once import QueueOnce

celery_app = Celery(
    broker=CELERY_BROKER_URL,
    backend=CELERY_BACKEND_URL
)
celery_app.conf.ONCE = {
  'backend': 'celery_once.backends.Redis',
  'settings': {
    'url': CELERY_BROKER_URL,
    'default_timeout': 60 * 60
  }
}
# 默认队列
default_exchange = Exchange(CELERY_DEFAULT_QUEUE, type='direct')
# 计算类任务队列，用于处理cpu密集型任务
compute_exchange = Exchange('compute', type='direct')
# eventlet协程队列，用于处理大并发量不耗cpu的任务，如爬虫等
eventlet_exchange = Exchange('eventlet', type='direct')
# 实时任务处理队列,用于处理kafka监听，binlog监听等实时监听任务
realtime_exchange = Exchange('realtime', type='direct')

# 定义不同的队列
CELERY_QUEUES = (
    Queue(CELERY_DEFAULT_QUEUE, default_exchange, routing_key=CELERY_DEFAULT_QUEUE),
    Queue('compute', compute_exchange, routing_key='compute'),
    Queue('eventlet', eventlet_exchange, routing_key='eventlet'),
    Queue('realtime', realtime_exchange, routing_key='realtime')
)


celery_app.conf.update(
    CELERY_DEFAULT_QUEUE=CELERY_DEFAULT_QUEUE,
    CELERY_DEFAULT_EXCHANGE=CELERY_DEFAULT_QUEUE,
    CELERY_DEFAULT_ROUTTING_KEY=CELERY_DEFAULT_QUEUE,
    CELERY_TASK_SERIALIZER='json',
    CELERY_ACCEPT_CONTENT=['json'],  # Ignore other content
    CELERY_RESULT_SERIALIZER='json',
    CELERY_ENABLE_UTC=True,
    CELERY_TASK_PROTOCOL=1,
    CELERYD_FORCE_EXECV=True,    # 有些情况下可以防止死锁
    CELERYD_MAX_TASKS_PER_CHILD=500,    # 每个worker最多执行n个任务就会被销毁重新开启进程，可防止内存泄露
    # CELERYD_TASK_TIME_LIMIT=60 * 60 * 24,   # 单个任务的运行时间不超过此值，否则会被SIGKILL 信号杀死
    # CELERY_IGNORE_RESULT=False,  # 是否忽略结果,
    # 任务结果保留时间
    CELERY_TASK_RESULT_EXPIRES=60 * 60
)


class MyTask(QueueOnce):

    def before_start(self, task_id, args, kwargs):
        '''
        任务开始前
        :param task_id:
        :param args:
        :param kwargs:
        :return:
        '''
        pass