Celery version 4.0 runs on
        Python ❨2.7, 3.4, 3.5❩
        PyPy ❨5.4, 5.5❩
    This is the last version to support Python 2.7, and from the next version (Celery 5.x) Python 3.5 or newer is required.

    If you're running an older version of Python, you need to be running an older version of Celery:

        Python 2.6: Celery series 3.1 or earlier.
        Python 2.5: Celery series 3.0 or earlier.
        Python 2.4 was Celery series 2.2 or earlier.

    Celery is a project with minimal funding, so we don't support Microsoft Windows. Please don't open any issues related to that platform.

from celery import Celery

# 第一個參數就是當前腳本的名稱，backend 任務執行結果的存儲地址broker 任務隊列的存儲地址
app = Celery('tasks', backend='redis://127.0.0.1', broker='redis://127.0.0.1')

@app.task
def add(x, y):
    return x + y

from tasks import add

result = add.delay(1, 2)
print('Is task ready: %s' % result.ready())  # False說明任務還沒有執行完
run_result = result.get(timeout=1)
print('task result: %s' % run_result)

print('Is task ready: %s' % result.ready())

$ cd celry
$ celery -A tasks worker --loglevel=info

cd celery
celery -A tasks flower --broker=redis://@localhost:6379/0

import os

from celery import Celery

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SEttINGS_MODULE', 'recruitment.base')  # 這里我把配置文件放到了根目錄下的settings/base.py 中

app = Celery('recruitment')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django apps.
app.autodiscover_tasks()

def debug_task(self):
  print(f'Request: {self.request!r}')

from __future__ import absolute_import, unicode_literals


# This will make sure the app is always imported when/保證所有app下的任務都能導入進來
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ('celery_app',)

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERYD_MAX_TASKS_PER_CHILD = 10
CELERYD_LOG_FILE = os.path.join(BASE_DIR, "logs", "celery_work.log")
CELERYBEAT_LOG_FILE = os.path.join(BASE_DIR, "logs", "celery_beat.log")

from __future__ import absolute_import, unicode_literals

from celery import shared_task
from .dingtalk import send

@shared_task
def send_dingtalk_message(message):
    send(message)

from dingtalkchatbot.chatbot import DingtalkChatbot

from django.conf import settings


def send(message, at_mobiles=[]):
    # 引用 settings里面配置的釘釘群消息通知的WebHook地址:
    webhook = settings.DINGTALK_WEB_HOOK

    # 初始化機器人小Y，
    xiaoY = DingtalkChatbot(webhook)

    # 方式二：勾選“加簽”選項時使用（v1.5以上新功能）
    # xiaoY = DingtalkChatbot(webhook, secret=secret)

    # Text消息@所有人
    xiaoY.send_text(msg=('消息通知: %s' % message), at_mobiles=at_mobiles)

from interview.tasks import send_dingtalk_message

def notify_interview(modeladmin, request, queryset):
    candidates = ''
    interviewers = ''
    for obj in queryset:
        candidates = obj.userame + '' + candidates
        interviewers = obj.first_interviewer_user + '' + interviewers
    # 這里的消息發送到釘釘， 或者通過 Celery 異步發送到釘釘
    send_dingtalk_message.delay('候選人 %s 進入面試環節， 親愛的面試官請做好面試準備：%s。' % (candidates, interviewers))

$ celery -A recruitment worker -l info

$ DJANGO_SEttINGS_MODULE=settings.base celery --app=recruitment worker --loglevel=info

$ celery -A recruitment flower --broker=redis://localhost:6379/0