目前演练的使用celery版本：5.0.5
> 配置模块

import yaml
from datetime import timedelta
from celery.schedules import crontab
from guigui_app.utils.common import DEFAULT_CONFIG_PATH
from kombu import Queue
from guigui_app.utils.typing_check import typeassert
from guigui_app.utils.web_jump_url import *
from guigui_app.utils.auto_import import auto_import_method

with open(DEFAULT_CONFIG_PATH, 'r', encoding='utf-8') as f:
    env_cfg = f.read()
env_setting = yaml.load(env_cfg, Loader=yaml.FullLoader)
redis_setting = env_setting.get('redis')

broker_url = 'redis://:{password}@{host}:{port}/{broker_db}'.format_map(redis_setting)
result_backend = 'redis://:{password}@{host}:{port}/{backend_db}'.format_map(redis_setting)
imports = ['celery_task.celery_task', ]  # 指定导入的任务模块
accept_content = ['application/json']
timezone = 'Asia/Shanghai'
# 并行数量 (进程数量)
worker_concurrency = 4
# 防止死锁
celeryd_force_execv = True
task_reject_on_worker_lost = True
# 预取任务数
worker_prefetch_multiplier = 1
result_expires = 60 * 60
task_default_queue = 'celery_new_auto_save_order'
# 任务队列

task_queues = (
    # 路由键以“celery_new_auto_save_order.”开头的消息都进default队列
    Queue("default", routing_key="celery_new_auto_save_order.#"),
    # 路由键以“auto.”开头的消息都进tasks_auto队列
    Queue("tasks_auto", routing_key="auto.#")
)

task_routes = (
    [  # 将celery_new_auto_save_order任务分配至队列 default
        ("celery_task.celery_task.celery_new_auto_save_order", {"queue": "default", "priority": 0}),
        # 将tasks_auto任务分配至队列 tasks_auto
        ("celery_task.celery_task.tasks_auto", {"queue": "tasks_auto", "priority": 3})
    ],
)

# 定时任务config
beat_schedule = {
    'celery_task.test': {
        'task': 'celery_task.celery_task.tasks_auto',
        'schedule': 3,
        'args': ('guigui_app.test', 'test', [{'a': 1, 'b': 2}])
    },
    'celery_task.add': {
        'task': 'celery_task.celery_task.tasks_auto',
        'schedule': 6,
        'args': ('celery_task.celery_task', 'add', [{'a': 1, 'b': 2}])

    },
}

> 应用模块

import logging
from celery import Celery
from guigui_app.utils.typing_check import typeassert
from guigui_app.utils.auto_import import auto_import_method

# 实例化celery 对象
app = Celery('tools_server')
# 导入celery 配置
app.config_from_object('celery_task.celery_settings')

logger = logging.getLogger('celery_auto_import')

@app.task
@typeassert(str, str, list)
def tasks_auto(mod: str, method_name: str, method_param: list = None):
    result = auto_import_method(mod, method_name, method_param=method_param)
    return result

 - timedelta是datetime中的一个对象，需要from datetime import timedelta引入，有如下几个参数
    days：天
    seconds：秒
    microseconds：微妙
    milliseconds：毫秒
    minutes：分
    hours：小时
- crontab的参数有：
    month_of_year：月份
    day_of_month：日期
    day_of_week：周
    hour：小时
    minute：分钟

import logging

log = logging.getLogger('auto_import')

__all__ = ['auto_import_method']


def auto_import_method(mod, method_name, method_param: list = None):
    obj = __import__(mod, fromlist=True)
    if hasattr(obj, method_name):
        log.info(f"Start path:{str(mod)} method: {str(method_name)}")
        func = getattr(obj, method_name)
        if method_param is None:
            return func()
        else:
            if isinstance(method_param, list):
                return func(method_param)
            else:
                log.warning(f"Error method param: {str(method_param)} method param type must be list")
                raise TypeError(f"Error method param: {str(method_param)} method param type must be list")
    else:
        log.error(f"error method {str(mod)} {str(method_name)}")
        raise NameError(f"There is no attr {method_name}  in {str(mod)}.py")

from inspect import signature
from functools import wraps


def typeassert(*ty_args, **ty_kwargs):
    def decorate(func):

        sig = signature(func)
        bund_types = sig.bind_partial(*ty_args, **ty_kwargs).arguments

        @wraps(func)
        def wrapper(*args, **kwargs):
            bund_values = sig.bind(*args, **kwargs)
            for name, value in bund_values.arguments.items():
                if name in bund_types:
                    if not isinstance(value, bund_types[name]):
                        raise TypeError('Argument {} must be {}'.format(name, bund_types[name]))
            return func(*args, **kwargs)

        return wrapper

    return decorate

方式一
# 启动定时任务
# celery -A celery_task.celery_task beat --loglevel=INFO
# 启动worker
# celery -A celery_task.celery_task worker --loglevel=INFO

方式二
以守护进程的方式启动
- worker
celery multi stop 6 -A celery_task.celery_task --loglevel=INFO --pidfile=/var/run/celery/%n.pid --logfile=/var/log/celery/%n%I.log
- beat
celery -A celery_task.celery_task beat --detach -l info -f beat.log

方式三
一起启动 
celery -A celery_task.celery_task worker --concurrency=6 --loglevel=INFO -B

tasks_auto.apply_async(args=(
    'guigui_app.celery_task.add_or_update_room_gltf_data', 'start_gltf_url',
    [rid_str]),
    countdown=1)
或者
tasks_auto.delay(
    'guigui_app.celery_task.add_or_update_room_gltf_data', 'start_gltf_url',
    [rid_str])

# Celery在执行任务时候，提供了钩子方法用于在任务执行完成时候进行对应的操作
# 在Task源码中提供了很多状态钩子函数如:
#   on_success(成功后执行)
#   on_failure(失败时候执行)
#   on_retry(任务重试时候执行)
#   after_return(任务返回时候执行)
# 在进程中使用是我们只需要重写这些方法，完成相应的操作即可

from celery import Task

logger = get_task_logger(__name__)

class demotask(Task):
    # 任务成功执行
    def on_success(self, retval, task_id, *args, **kwargs):
        logger.info(f'task id:{task_id}, arg:{args}, successful!')

    # 任务失败执行
    def on_failure(self, exc, task_id, *args, **kwargs, einfo):
        logger.info(f'task id:{task_id}, arg:{args}, failed! erros:{exc}')

    # 任务重试执行
    def on_retry(self, exc, task_id, *args, **kwargs, einfo):
        logger.info(f'task id:{task_id}, arg:{args}, retry! einfo:{exc}')

@app.task(base=demotask)

# --------------------------------------------
# delay源码
# --------------------------------------------
def delay(self, *args, **kwargs):
        """Star argument version of :meth:`apply_async`.

        Does not support the extra options enabled by :meth:`apply_async`.

        Arguments:
            *args (Any): Positional arguments passed on to the task.
            **kwargs (Any): Keyword arguments passed on to the task.
        Returns:
            celery.result.AsyncResult: Future promise.
        """
        return self.apply_async(args, kwargs)
# --------------------------------------------
# apply_async源码
# --------------------------------------------
def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
                    link=None, link_error=None, shadow=None, **options):
        if self.typing:
            try:
                check_arguments = self.__header__
            except AttributeError:  # pragma: no cover
                pass
            else:
                check_arguments(*(args or ()), **(kwargs or {}))

        app = self._get_app()
        if app.conf.task_always_eager:
            with denied_join_result():
                return self.apply(args, kwargs, task_id=task_id or uuid(),
                                  link=link, link_error=link_error, **options)

        if self.__v2_compat__:
            shadow = shadow or self.shadow_name(self(), args, kwargs, options)
        else:
            shadow = shadow or self.shadow_name(args, kwargs, options)

        preopts = self._get_exec_options()
        options = dict(preopts, **options) if options else preopts

        options.setdefault('ignore_result', self.ignore_result)

        return app.send_task(
            self.name, args, kwargs, task_id=task_id, producer=producer,
            link=link, link_error=link_error, result_cls=self.AsyncResult,
            shadow=shadow, task_type=self,
            **options
        )