项目文件：

其中，tasks是消费者模块，有管理的celery文件，有多个任务函数文件。produce_task1模拟消费者，测试异步任务调用。

celery.py

from celery import Celery
from datetime import timedelta

app = Celery("tasks_demo",
             broker="redis://192.168.124.49:6379/0",
             backend="redis://192.168.124.49:6379/0",
             # 包含以下两个任务文件，去相应的py文件中找任务，对多个任务做分类
             include=["celery_demo2.tasks.task1", "celery_demo2.tasks.task2"])
app.conf.timezone = 'Asia/Shanghai'  # 时区
app.conf.enable_utc = False  # 是否使用UTC
app.conf.task_default_queue = "celery02"  # 修改默认队列,可以不要

# 配置文件定时任务
app.conf.beat_schedule = {
    'sendmail-every-10-seconds': {
        'task': 'celery_demo2.tasks.task2.send_msg',
        'schedule': timedelta(seconds=10),
        'args': ('李四',)
    },
}

task1.py

import time
from celery_demo2.tasks.celery import app


# 这是关键，穿上这件衣服就是异步任务函数了
@app.task
def send_email(res):
    for i in range(5):
        print(f"{i}开始向{res}发送邮件任务")
        time.sleep(3)
        print(f"{i}完成向{res}发送邮件任务")
    return "mail ok"

task2.py

import time
from celery_demo2.tasks.celery import app


@app.task
def send_msg(name):
    for i in range(5):
        print(f"{i}开始向{name}发送短信任务")
        time.sleep(3)
        print(f"{i}完成向{name}发送短信任务")
    return "msg ok"

produce_task1.py

from celery_demo2.tasks.task1 import send_email
from celery_demo2.tasks.task2 import send_msg
from celery_demo2.tasks.celery import app
from datetime import datetime
import time

v1 = datetime.now()
print(f"当前时间:{v1}")
result1 = send_email.delay('张三')
print(f"任务ID{result1.id}")
result2 = send_msg.delay('王五')
print(f"任务ID{result2.id}")

time.sleep(10)
app.control.revoke(result1.id, terminate=True)
print("程序执行完毕！")

启动celery：

celery -A celery_demo2.tasks.celery worker --loglevel=info -P eventlet

注意：在windows中执行时，添加-P eventlet，否则可能报错。

运行produce_task1.py文件，模拟生产者调用异步任务:

参考博文：

celery简单实现异步任务和定时任务

celery 停止执行中 task