import pymysql

pymysql.version_info=(1,4,3,"final",0) # 指定了pymysql的版本：1.4.3,按照版本修改

pymysql.install_as_MySQLdb()

from .celery import app as celery_app

__all__ = ['celery_app']

3.settings.py 配置

# Celery配置

# from kombu import Exchange, Queue

# 设置任务接受的类型，默认是{'json'}

CELERY_ACCEPT_CONTENT = ['application/json']

# 设置task任务序列列化为json

CELERY_TASK_SERIALIZER = 'json'

# 请任务接受后存储时的类型

CELERY_RESULT_SERIALIZER = 'json'

# 时间格式化为中国时间

CELERY_TIMEZONE = 'Asia/Shanghai'

# 是否使用UTC时间

CELERY_ENABLE_UTC = False

# 指定borker为redis 如果指定rabbitmq CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//'

CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'

# 指定存储结果的地方，支持使用rpc、数据库、redis等等，具体可参考文档

# CELERY_RESULT_BACKEND = 'db+mysql://scott:tiger@localhost/foo'

# mysql 作为后端数据库

CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1'

# 设置任务过期时间 默认是一天，为None或0 表示永不过期

CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24

# 设置worker并发数，默认是cpu核心数

# CELERYD_CONCURRENCY = 12

# 设置每个worker最大任务数

CELERYD_MAX_TASKS_PER_CHILD = 100

# 指定任务的位置

CELERY_IMPORTS = (

'base.tasks',

)

# 使用beat启动Celery定时任务

# schedule时间的具体设定参考：https://docs.celeryproject.org/en/stable/userguide/periodic-tasks.html

# 定时任务

CELERY_BEAT_SCHEDULE = {

"orders_notpay": { # 定时任务

'task': 'base.tasks.cancelorders',

'schedule': 3, # 定时任务的调用时间，10表示每隔10秒调用一次任务，5*60表示每5分钟

# "schedule": crontab(hour=7, minute=30, day_of_week=1),, # 定时任务的调用时间，每周一早上7点30分调用一次add任务

},

"syn_mysqlEs": { # 定时任务：每一分钟同步一次mysql到es的数据

'task': 'base.tasks.syn_mysql_es',

'schedule': 3, # 定时任务的调用时间，10表示每隔10秒调用一次任务，5*60表示每5分钟

# "schedule": crontab(hour=7, minute=30, day_of_week=1),, # 定时任务的调用时间，每周一早上7点30分调用一次add任务

},

}

4.设置tasks 任务：

#base/tasks.py

from celery import shared_task

from ronglianyunapi import send_sms as sms

# 记录日志：

import logging

logger = logging.getLogger("django")

@shared_task(name="send_sms")code>

def send_sms(tid, mobile, datas):

"""异步发送短信"""

try:

return sms(tid, mobile, datas)

except Exception as e:

logger.error(f"发送短信失败: { e}")

5.启动定时任务：

第一个终端：celery -A web_back worker -l INFO -P eventlet第二个终端：celery -A web_back beat

<code>@shared_task

def cancelorders():

# 获取订单号 0 小于等于int（time.time()）

# 遍历订单号列表

# 根据订单号查询订单表，把订单更新为失败状态

# 查看订单是否使用积分和优惠券，如果使用更新，写入积分记录表，用户表中的积分+，优惠券+

orderno = r.get_str("orderno")

order = OrdersModel.objects.filter(transaction=orderno).first()

pay_status = order.pay_status

if pay_status == 1: #生成订单-未支付状态

order.orders_status = 7 #订单-->取消状态

#...

print("取消订单操作")

@shared_task

def ordersuccess():

#读取成功的队列，获取到订单号

len = r.list_len("ordersuccess")

while len > 0:

sid = r.list_pop("ordersuccess")

# 根据订单号查询订单

order = OrdersDetailModel.objects.filter(orders_id=sid).first()

# 更新用户课程表

UserCourseModel.objects.create(user_id=order.user_id,course_id=order.course_id)

# 更新用户课程章节表

UserCourseChapterModel.objects.create(user_id=order.user_id,course_id=order.course_id,name=order.course.name)

print("订单成功时，更新用户课程表和用户课程章节表")

@shared_task

def orderfail():

# 读取失败的队列，获取到订单号，

len = r.list_len("orderfail")

while len > 0:

fid = r.list_pop("orderfail")

# 根据订单号查询订单

order = OrdersModel.objects.filter(orders_id=fid).first()

# 更新积分记录表、用户表中总积分、优惠券

srecord = ScoreRecordModel.objects.filter(user_id=order.user_id).first()

srecord.type = 1

srecord.score = order.score

srecord.save()

user = UsersModel.objects.filter(user_id=order.user_id).first()

user.points += order.score

userCoupon = UserCouponModel.objects.filter(user_id=order.user_id).first()

userCoupon.status = "未使用"

print("订单失败时，更新数据完成")