测试派 Give testers a home.

前言

上一篇文章给大家介绍了celery的基本使用，接下来给大家介绍一下在Django中如何使用celery来实现有任务和定时任务。

一、Django+celery实现异步任务

1、项目结构介绍

创建django项目celeryApp

项目中创建应用App01

├─App01
│  ├─admin.py
│  ├─view.py
│  ├─models.py
│  ├─apps.py
│  └─tasks.py
│ 
├─celeryApp
│  ├─__init__.py
│  ├─uwsgi.py
│  ├─urls.py
│  ├─settings.py
│  └─celery.py
├─manage.py

• 项目setting.py同级的目录创建下celery.py
• 应用目录中创建一个任务的模块tasks.py

2、编写celery.py

import os
from celery import Celery

# 创建celery实例对象
app = Celery("musen",
          broker='redis://redis服务host地址:6379/1',
          backend='redis://redis服务host地址:6379/2')

# 把celery和django进行组合，识别和加载django的配置文件
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celeryPros.settings.dev')

# 加载任务
# 参数必须必须是一个列表，列表里面上有异步任务的应用名称,celery会自动加载应用目录中的`tasks.py`的任务函数
# app.autodiscover_tasks(["应用1","应用1"])
app.autodiscover_tasks(["celeryApp"])

3、编写任务函数tasks.py

from celeryApp.celery import app
import time

import logging
log = logging.getLogger("django")

@app.task  
def send_sms(mobile):
	"""发送短信的异步任务"""
 print("小柠檬向手机号{}发送短信成功!".format(mobile))
 time.sleep(2)
 return "发送短信完成 OK"

4、启动celery

# 启动Celery的命令
celery -A celeryApp worker --loglevel=info -P threads

5、django视图函数中调用celery定义的异步任务

from django.shortcuts import HttpResponse
from datetime import datetime
from celeryApp.corn_task.tasks import send_sms

def run_test(request):
	"""执行celery任务的视图"""
 # 提交异步任务
 send_sms.delay('199')

 # 提交定时任务
 t = datetime(2022, 3, 4, 14, 41, 00)
 t = datetime.utcfromtimestamp(t.timestamp())
 send_sms.apply_async(args=["小联盟",], eta=t)

 return HttpResponse('异步任务已执行')

6、配置好路由，启动Django，访问视图对应的路由，就会看到celery执行异步任务。

二、django+celery实现定时任务

如果想要在django中使用定时任务功能同样是靠beat完成任务发送功能，当在Django中使用定时任务时，需要安装django-celery-beat插件。以下将介绍使用过程。

1、环境安装

pip install Django-celery-beat

2、任务编写和配置

1、注册django_celery_beat

在settings.py的INSTALLED_APPS中注册django_celery_beat

INSTALLED_APPS = [
  ...
   'django_celery_beat'

]

2、重新生成迁移

执行python manage.py migrate生成迁移

3、setting.py编写celery配置项

# =========celery的配置============
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60
# 任务队列配置
CELERY_BROKER_URL = 'redis://redis服务host地址:6379/1'

4、编写异步任务

• 1、项目setting.py同级的目录创建下celery.py，(和前面django+celery实现异步任务一样)

  import os
  from celery import Celery
  
  # 把celery和django进行组合，识别和加载django的配置文件
  os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celeryPros.settings.dev')
  # 创建celery实例对象
  app = Celery('CeleryDjango')
  # 加载setting.py中django的配置
  app.config_from_object('django.conf:settings', namespace='CELERY')
  
  # 加载任务
  app.autodiscover_tasks(["App01"])
  

• 2、应用目录中创建一个任务的模块tasks.py，编写异步任务

  from celeryApp.celery import app
  import time
  
  import logging
  log = logging.getLogger("django")
  
  @app.task  
  def test_run(datas):
  	"""运行测试的定时任务"""
   print("开始执行测试用例,用例数据为{}".format(datas))
   time.sleep(2)
   return "用例执行完成"
  

3、启动celery服务

1、启动celery worker服务

celery -A 项目名 worker --loglevel=info -P threads

2、启动celery beat服务

$ celery -A 项目名 beat -l INFO --scheduler django_celery_beat.schedulers:DatabaseScheduler

4、添加定时任务

1、登录admin后台管理系统

启动django服务，登录amdin后台管理系统,可以看到如下几张表，

periodic tasks为定时任务，其他的为定时任务的定时模式，

• Clocked模式 : 可以设置一个固定的时间来执行任务。
• Crontabs模式： 可以设置任务在一天中的特定时间或一周中的某一天 的某个时刻周期性执行任务。
• intervals模式：可以设置任务安装固定的间隔时间来执行。
• solar events模式：可以设置任务 按照某个经纬度的日出、日落、黎明或黄昏等时间来周期性的执行任务。

创建定时任务时，需选择以上以上四种中的一种，来定时执行任务。

2、创建定时任务