詳解分布式任務(wù)隊(duì)列Celery使用說(shuō)明

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')

# 通過(guò) Celery 實(shí)例加載配置模塊
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'

# 導(dǎo)入指定的任務(wù)模塊
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()   # 查看任務(wù)狀態(tài)，返回布爾值, 任務(wù)執(zhí)行完成, 返回 True, 否則返回 False.
r.wait()   # 等待任務(wù)完成, 返回任務(wù)執(zhí)行結(jié)果，很少使用；
r.get(timeout=1)    # 獲取任務(wù)執(zhí)行結(jié)果，可以設(shè)置等待時(shí)間
r.result   # 任務(wù)執(zhí)行結(jié)果.
r.state    # PENDING, START, SUCCESS，任務(wù)當(dāng)前的狀態(tài)
r.status   # PENDING, START, SUCCESS，任務(wù)當(dāng)前的狀態(tài)
r.successful # 任務(wù)成功返回true
r.traceback # 如果任務(wù)拋出了一個(gè)異常，你也可以獲取原始的回溯信息

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) # 第一個(gè)參數(shù)是self，使用self.request訪問(wèn)相關(guān)的屬性
def add(self, x, y):
  logger.info(self.request.id)

import celery

class MyTask(celery.Task):
  # 任務(wù)失敗時(shí)執(zhí)行
  def on_failure(self, exc, task_id, args, kwargs, einfo):
    print('{0!r} failed: {1!r}'.format(task_id, exc))
  # 任務(wù)成功時(shí)執(zhí)行
  def on_success(self, retval, task_id, args, kwargs):
    pass
  # 任務(wù)重試時(shí)執(zhí)行
  def on_retry(self, exc, task_id, args, kwargs, einfo):
    pass

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

exc:失敗時(shí)的錯(cuò)誤的類型；
task_id:任務(wù)的id；
args:任務(wù)函數(shù)的參數(shù)；
kwargs:參數(shù)；
einfo:失敗時(shí)的異常詳細(xì)信息；
retval:任務(wù)成功執(zhí)行的返回值；

# 注意，celery4版本后，CELERY_BROKER_URL改為BROKER_URL
BROKER_URL = 'amqp://username:passwd@host:port/虛擬主機(jī)名'
# 指定結(jié)果的接受地址
CELERY_RESULT_BACKEND = 'redis://username:passwd@host:port/db'
# 指定任務(wù)序列化方式
CELERY_TASK_SERIALIZER = 'msgpack' 
# 指定結(jié)果序列化方式
CELERY_RESULT_SERIALIZER = 'msgpack'
# 任務(wù)過(guò)期時(shí)間,celery任務(wù)執(zhí)行結(jié)果的超時(shí)時(shí)間
CELERY_TASK_RESULT_EXPIRES = 60 * 20  
# 指定任務(wù)接受的序列化類型.
CELERY_ACCEPT_CONTENT = ["msgpack"]  
# 任務(wù)發(fā)送完成是否需要確認(rèn)，這一項(xiàng)對(duì)性能有一點(diǎn)影響   
CELERY_ACKS_LATE = True 
# 壓縮方案選擇，可以是zlib, bzip2，默認(rèn)是發(fā)送沒(méi)有壓縮的數(shù)據(jù)
CELERY_MESSAGE_COMPRESSION = 'zlib' 
# 規(guī)定完成任務(wù)的時(shí)間
CELERYD_TASK_TIME_LIMIT = 5 # 在5s內(nèi)完成任務(wù)，否則執(zhí)行該任務(wù)的worker將被殺死，任務(wù)移交給父進(jìn)程
# celery worker的并發(fā)數(shù)，默認(rèn)是服務(wù)器的內(nèi)核數(shù)目,也是命令行-c參數(shù)指定的數(shù)目
CELERYD_CONCURRENCY = 4 
# celery worker 每次去rabbitmq預(yù)取任務(wù)的數(shù)量
CELERYD_PREFETCH_MULTIPLIER = 4 
# 每個(gè)worker執(zhí)行了多少任務(wù)就會(huì)死掉，默認(rèn)是無(wú)限的
CELERYD_MAX_TASKS_PER_CHILD = 40 
# 這是使用了django-celery默認(rèn)的數(shù)據(jù)庫(kù)調(diào)度模型,任務(wù)執(zhí)行周期都被存在你指定的orm數(shù)據(jù)庫(kù)中
# CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'
# 設(shè)置默認(rèn)的隊(duì)列名稱，如果一個(gè)消息不符合其他的隊(duì)列就會(huì)放在默認(rèn)隊(duì)列里面，如果什么都不設(shè)置的話，數(shù)據(jù)都會(huì)發(fā)送到默認(rèn)的隊(duì)列中
CELERY_DEFAULT_QUEUE = "default"
# 設(shè)置詳細(xì)的隊(duì)列
CELERY_QUEUES = {
  "default": { # 這是上面指定的默認(rèn)隊(duì)列
    "exchange": "default",
    "exchange_type": "direct",
    "routing_key": "default"
  },
  "topicqueue": { # 這是一個(gè)topic隊(duì)列 凡是topictest開頭的routing key都會(huì)被放到這個(gè)隊(duì)列
    "routing_key": "topic.#",
    "exchange": "topic_exchange",
    "exchange_type": "topic",
  },
  "task_eeg": { # 設(shè)置扇形交換機(jī)
    "exchange": "tasks",
    "exchange_type": "fanout",
    "binding_key": "tasks",
  },

}