项目文件：
其中，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文件，模拟生产者调用异步任务:
参考博文：
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/142829.html
