详解分布式任务队列Celery使用说明

pip install celery

pip install redis

# tasks.py
import time
from celery import Celery

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery('my_tasks', broker=broker, backend=backend)

@app.task
def add(x, y):
  print('enter task')
  time.sleep(3)
  return x + y

# app.py
from tasks import add

if __name__ == '__main__':
  print('start task')
  result = add.delay(2, 18)
  print('end task')
  print(result)

celery worker -A tasks -l info  # -A 后是模块名

pip install eventlet
celery -A <mymodule> worker -l info -P eventlet

import os
os.environ.setdefault('FORKED_BY_MULTIPROCESSING', '1')

# __init__.py
import os
from celery import Celery

os.environ.setdefault('FORKED_BY_MULTIPROCESSING', '1')
app = Celery('demo')

# 通过 Celery 实例加载配置模块
app.config_from_object('celery_app.celery_config')

# celery_config.py
BROKER_URL = 'redis://127.0.0.1:6379/1'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/2'

# UTC
CELERY_ENABLE_UTC = True
CELERY_TIMEZONE = 'Asia/Shanghai'

# 导入指定的任务模块
CELERY_IMPORTS = (
  'celery_app.task1',
  'celery_app.task2',
)

# task1.py
import time
from celery_app import app

@app.task
def add(x, y):
  print('enter task')
  time.sleep(3)
  return x + y

# task2.py
import time
from celery_app import app

@app.task
def mul(x, y):
  print('enter task')
  time.sleep(4)
  return x * y

# app.py
from celery_app import task1

if __name__ == '__main__':
  pass
  print('start task')
  result = task1.add.delay(2, 18)
  print('end task')
  print(result)

$ python app.py
$ celery worker -A celery_app -l info

r.ready()   # 查看任务状态，返回布尔值, 任务执行完成, 返回 True, 否则返回 False.
r.wait()   # 等待任务完成, 返回任务执行结果，很少使用；
r.get(timeout=1)    # 获取任务执行结果，可以设置等待时间
r.result   # 任务执行结果.
r.state    # PENDING, START, SUCCESS，任务当前的状态
r.status   # PENDING, START, SUCCESS，任务当前的状态
r.successful # 任务成功返回true
r.traceback # 如果任务抛出了一个异常，你也可以获取原始的回溯信息

import datetime
from celery.schedules import crontab

CELERYBEAT_SCHEDULE = {
  'task1-every-1-min': {
    'task': 'celery_app.task1.add',
    'schedule': datetime.timedelta(seconds=60),
    'args': (2, 15),
  },
  'task2-once-a-day': {
    'task': 'celery_app.task2.mul',
    'schedule': crontab(hour=15, minute=23),
    'args': (3, 6),
  }
}

$ celery worker -A celery_app -l info

$ celery -B -A celery_app worker -l info

@celery.task()
def name():
  pass

@task(bind=True) # 第一个参数是self，使用self.request访问相关的属性
def add(self, x, y):
  logger.info(self.request.id)

import celery

class MyTask(celery.Task):
  # 任务失败时执行
  def on_failure(self, exc, task_id, args, kwargs, einfo):
    print('{0!r} failed: {1!r}'.format(task_id, exc))
  # 任务成功时执行
  def on_success(self, retval, task_id, args, kwargs):
    pass
  # 任务重试时执行
  def on_retry(self, exc, task_id, args, kwargs, einfo):
    pass

@task(base=MyTask)
def add(x, y):
  raise KeyError()

exc:失败时的错误的类型；
task_id:任务的id；
args:任务函数的参数；
kwargs:参数；
einfo:失败时的异常详细信息；
retval:任务成功执行的返回值；

# 注意，celery4版本后，CELERY_BROKER_URL改为BROKER_URL
BROKER_URL = 'amqp://username:passwd@host:port/虚拟主机名'
# 指定结果的接受地址
CELERY_RESULT_BACKEND = 'redis://username:passwd@host:port/db'
# 指定任务序列化方式
CELERY_TASK_SERIALIZER = 'msgpack' 
# 指定结果序列化方式
CELERY_RESULT_SERIALIZER = 'msgpack'
# 任务过期时间,celery任务执行结果的超时时间
CELERY_TASK_RESULT_EXPIRES = 60 * 20  
# 指定任务接受的序列化类型.
CELERY_ACCEPT_CONTENT = ["msgpack"]  
# 任务发送完成是否需要确认，这一项对性能有一点影响   
CELERY_ACKS_LATE = True 
# 压缩方案选择，可以是zlib, bzip2，默认是发送没有压缩的数据
CELERY_MESSAGE_COMPRESSION = 'zlib' 
# 规定完成任务的时间
CELERYD_TASK_TIME_LIMIT = 5 # 在5s内完成任务，否则执行该任务的worker将被杀死，任务移交给父进程
# celery worker的并发数，默认是服务器的内核数目,也是命令行-c参数指定的数目
CELERYD_CONCURRENCY = 4 
# celery worker 每次去rabbitmq预取任务的数量
CELERYD_PREFETCH_MULTIPLIER = 4 
# 每个worker执行了多少任务就会死掉，默认是无限的
CELERYD_MAX_TASKS_PER_CHILD = 40 
# 这是使用了django-celery默认的数据库调度模型,任务执行周期都被存在你指定的orm数据库中
# CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'
# 设置默认的队列名称，如果一个消息不符合其他的队列就会放在默认队列里面，如果什么都不设置的话，数据都会发送到默认的队列中
CELERY_DEFAULT_QUEUE = "default"
# 设置详细的队列
CELERY_QUEUES = {
  "default": { # 这是上面指定的默认队列
    "exchange": "default",
    "exchange_type": "direct",
    "routing_key": "default"
  },
  "topicqueue": { # 这是一个topic队列 凡是topictest开头的routing key都会被放到这个队列
    "routing_key": "topic.#",
    "exchange": "topic_exchange",
    "exchange_type": "topic",
  },
  "task_eeg": { # 设置扇形交换机
    "exchange": "tasks",
    "exchange_type": "fanout",
    "binding_key": "tasks",
  },

}