网站建设服务器环境配置,制作网站的软件下载,聊城推广网站,wordpress 早期目录 一、依赖包情况
二、项目目录结构 2.1、怎么将django的应用创建到apps包
三、celery的配置
2.1、celery_task/celery.py
2.2、celery_task/async_task.py 2.3、celery_task/scheduler_task.py
2.4、utils/check_task.py
四、apps/user中配置相关处理视图
4.1、基本…目录 一、依赖包情况
二、项目目录结构 2.1、怎么将django的应用创建到apps包
三、celery的配置
2.1、celery_task/celery.py
2.2、celery_task/async_task.py 2.3、celery_task/scheduler_task.py
2.4、utils/check_task.py
四、apps/user中配置相关处理视图
4.1、基本配置
4.2、user的models
4.3、user的视图函数
五、调用函数测试
5.1、启动项目
5.2、测试项目Postman接口工具
六、报错 一、依赖包情况 python3.9.0 django3.2.0 celery5.3.1 django-redis5.3.0 eventlet0.33.3 #windows系统需要使用到 注意还需要在系统中安装好redis数据库不然无法使用。 二、项目目录结构 2.1、怎么将django的应用创建到apps包
·1、创建user应用
cd apps
python ../manage.py startapp user 2、修改user包下的apps.py模块
from django.apps import AppConfigclass UserConfig(AppConfig):default_auto_field django.db.models.BigAutoField#原来是 name user,改成下面的name apps.user3、注册到settings.py文件中
INSTALLED_APPS [django.contrib.admin,django.contrib.auth,django.contrib.contenttypes,django.contrib.sessions,django.contrib.messages,django.contrib.staticfiles,apps.user.apps.UserConfig, #注册user应用from apps.user.apps import UserConfig
]
三、celery的配置
概述celery应用程序和任务都放到celery_task包中其中celery.py是创建Celery的实例对象的async_task.py用来写异步任务scheduler_task.py用来写定时任务的。utils/check_task.py用来检测任务id是否结束并获取任务的返回值的。 2.1、celery_task/celery.py
from celery import Celery
from celery.schedules import crontab
from datetime import timedelta# 消息中间件密码是你redis的密码
# brokerredis://:123456127.0.0.1:6379/2 密码123456
broker redis://127.0.0.1:6379/0 # 无密码# 任务结果存储
backend redis://127.0.0.1:6379/1#包含任务的所有模块的导入路径
task_module [celery_task.async_task, #写任务模块导入路径该模块主要写异步任务的方法celery_task.scheduler_task, #写任务模块导入路径该模块主要写定时任务的方法
]# 生成celery对象task相当于key用于区分celery对象
# broker是指定消息处理backend是指定结果后端的存储位置 include参数需要指定任务模块
app Celery(task, brokerbroker, backendbackend, includetask_module)# 配置
# 时区
app.conf.timezone Asia/Shanghai
# 是否使用UTC
app.conf.enable_utc False
# 定时任务配置
app.conf.beat_schedule {# 名字随意命名add-func-30-seconds: {# 执行add_task下的addy函数task: celery_task.scheduler_task.add_func, # 任务函数的导入路径from celery_task.scheduler_task import add_func# 每10秒执行一次schedule: timedelta(seconds30),# add函数传递的参数args: (10, 21)},#名字随意起add-func-5-minutes: {task: celery_task.scheduler_task.add_func,# 任务函数的导入路径from celery_task.scheduler_task import add_func# crontab不传的参数默认就是每的意思比如这里是每年每月每日每天每小时的5分执行该任务schedule: crontab(minute5), # 之前时间点执行每小时的第5分钟执行任务, 改成小时分钟秒 就是每天的哪个小时哪分钟哪秒钟执行args: (19, 22) #定时任务需要的参数},#缓存用户数据到cache中cache-user-func:{task:celery_task.scheduler_task.cache_user_func,#导入任务函数from celery_task.scheduler_task import cache_user_funcschedule:timedelta(minutes1),#每1分钟执行一次将用户消息缓存到cache中}
}
配置也可以使用下面这种方式
app.conf.update(task_serializerjson,accept_content[json], # Ignore other contentresult_serializerjson,timezoneAsia/Shanghai,enable_utcFalse,
)2.2、celery_task/async_task.py
# 因为需要用到django中的内容所以需要配置django环境
import os
os.environ.setdefault(DJANGO_SETTINGS_MODULE, study_celery.settings)
import django
django.setup()# 导入celery对象app
from celery_task.celery import app
# 导入django自带的发送邮件模块
from django.core.mail import send_mail
import threading
from study_celery import settings1、没有返回值的app.task(ignore_resultTrue)
2、有返回值的任务app.task
#没有返回值禁用掉结果后端
app.task
def send_email_task(email,code): # 此时可以直接传邮箱还能减少一次数据库的IO操作:param email: 接收消息的邮箱用户的邮箱:return:# 启用线程发送邮件此处最好加线程池t threading.Thread(targetsend_mail,args(登录前获取的验证码, # 邮件标题点击该邮件激活你的账号否则无法登陆, # 给html_message参数传值后该参数信息失效settings.EMAIL_HOST_USER, # 用于发送邮件的邮箱地址[email], # 接收邮件的邮件地址可以写多个),# html_message中定义的字符串即HTML格式的信息可以在一个html文件中写好复制出来放在该字符串中kwargs{html_message: fp/p p验证码{code}/p})t.start()return {email:email,code:code}2.3、celery_task/scheduler_task.py
# 因为需要用到django中的内容所以需要配置django环境
import os
os.environ.setdefault(DJANGO_SETTINGS_MODULE, study_celery.settings)
import django
django.setup()from celery_task.celery import app
from apps.user.views import models as user_models
from django.core.cache import cache
import time
from django.forms import model_to_dict#有返回值返回值可以从结果后端中获取
app.task
def add_func(a,b):print(执行了加法函数)cache.set(add_ret,{time:time.strftime(%Y-%m-%d %H:%M:%S),ret:ab})return ab#不需要返回值禁用掉结果后端
app.task(ignore_resultTrue)
def cache_user_func():user user_models.UserModel.objects.all()user_dict {}for obj in user:user_dict[obj.account] model_to_dict(obj)cache.set(all-user-data,user_dict,timeout35*60)
2.4、utils/check_task.py
from celery.result import AsyncResult
from celery_task.celery import app
验证任务的执行状态的def check_task_status(task_id):任务的执行状态PENDING :等待执行STARTED :开始执行RETRY :重新尝试执行SUCCESS :执行成功FAILURE :执行失败:param task_id::return:result AsyncResult(idtask_id, appapp)dic {type:result.status,msg:,data:,code:400}if result.status PENDING:dic[msg] 任务等待中elif result.status STARTED:dic[msg] 任务开始执行elif result.status RETRY:dic[msg]任务重新尝试执行elif result.status FAILURE:dic[msg] 任务执行失败了elif result.status SUCCESS:result result.get()dic[msg] 任务执行成功dic[data] resultdic[code] 200# result.forget() # 将结果删除# async.revoke(terminateTrue) # 无论现在是什么时候都要终止# async.revoke(terminateFalse) # 如果任务还没有开始执行呢那么就可以终止。return dic 四、apps/user中配置相关处理视图
4.1、基本配置
1、study_celery/settings.py
#cache缓存
CACHES {default: {BACKEND: django_redis.cache.RedisCache,LOCATION: redis://127.0.0.1:6379,OPTIONS: {CLIENT_CLASS: django_redis.client.DefaultClient,CONNECTION_POOL_KWARGS: {max_connections: 100}# PASSWORD: 123,},TIMEOUT:30*60 #缓存过期时间}
}#邮件配置
# EMAIL_BACKEND django.core.mail.backends.smtp.EmailBackend
EMAIL_HOST smtp.qq.com # 如果是 163 改成 smtp.163.com
EMAIL_PORT 465
EMAIL_HOST_USER 2414155342qq.com # 发送邮件的邮箱帐号
EMAIL_HOST_PASSWORD qq邮箱的授权码 # 授权码,各邮箱的设置中启用smtp服务时获取
DEFAULT_FROM_EMAIL EMAIL_HOST_USER
# 这样收到的邮件收件人处就会这样显示
# DEFAULT_FROM_EMAIL xxxxxqq.com
EMAIL_USE_SSL True # 使用ssl
# EMAIL_USE_TLS False # 使用tls
# EMAIL_USE_SSL 和 EMAIL_USE_TLS 是互斥的即只能有一个为 True
2、study_celery/urls.py
from django.contrib import admin
from django.urls import path,includeurlpatterns [path(admin/, admin.site.urls),path(user/,include(apps.user.urls))
]
3、apps/user/urls.py
from django.urls import path
from . import viewsurlpatterns []
4.2、user的models
apps/user/models.py
from django.db import models
# Create your models here.class UserModel(models.Model):id models.AutoField(primary_keyTrue)name models.CharField(max_length32)account models.CharField(max_length64)password models.CharField(max_length256)email models.EmailField()
执行数据库迁移命令 python manage.py makemigrations python manage.py migrate 4.3、user的视图函数
1、apps/user/views.py
from django.contrib.auth.hashers import make_password, check_password
from django.views import View
from django.http import JsonResponse
from . import models
from django.core.cache import cache
import time
from celery_task.async_task import send_email_task
# Create your views here.class ResgiterView(View):#注册用户def post(self,request):name request.POST.get(name)account request.POST.get(account)password request.POST.get(password)email request.POST.get(email)obj models.UserModel.objects.filter(accountaccount).first()if obj:return JsonResponse({code:400,msg:账户已经存在了})password make_password(password)instance models.UserModel.objects.create(namename,accountaccount,passwordpassword,emailemail)return JsonResponse({code:200,msg:注册用户成功})class LoginView(View):#用户登录def post(self,request):account request.POST.get(account)password request.POST.get(password)code request.POST.get(code) #验证码email_code cache.get(femail_{account})#发给邮箱的验证码.get(femail_{account})print(code,email_code)if code and email_code:if code ! email_code:return JsonResponse({code:400,msg:验证码错误})else:return JsonResponse({code:400,msg:请先点击发送邮件获取验证码})obj models.UserModel.objects.filter(accountaccount).first()if not obj:return JsonResponse({code:400,msg:当前用户不存在})pwd_true check_password(password,obj.password)response JsonResponse({code:200,msg:登录成功})if pwd_true:return responseelse:return JsonResponse({code:400,msg:用户名或密码错误})class LoginSendEmailView(View):#用户登录前需要验证码发送验证码给用户的邮箱def post(self,request):account request.POST.get(account)email request.POST.get(email)code str(time.time())[-5:]cache.set(femail_{account},code)print(cache.get(femail_{account}))res send_email_task.delay(email,code)task_id res.idprint(验证码是,code)return JsonResponse({code:200,msg:f请查看{email}邮箱中是否收到邮件,task_id:task_id})class AllUserDataView(View):#查询cache_user_func定时任务执行时存到cache中的用户数据def get(self,request):key all-user-datadata cache.get(key)if data:return JsonResponse({code:200,data:data})else:return JsonResponse({code:400,msg:没有相关数据})class AddFuncDataView(View):#查询add_func 定时任务执行时存到cache中的数据def get(self, request):data cache.get(add_ret)print(data, type(data))return JsonResponse({code: 200, data: data})class UserTaskIdGetDataView(View):def get(self,request):from utils.check_task import check_task_status#检查任务是否成功了获取任务的返回值task_id request.GET.get(task_id)if not task_id:return JsonResponse({code:400,msg:没有携带任务id})ret check_task_status(task_id)return JsonResponse(ret)
2、apps/user/urls.py
from django.urls import path
from . import viewsurlpatterns [path(login/,views.LoginView.as_view(),nameuser-login),#登录path(register/, views.ResgiterView.as_view(), nameuser-register),#注册path(login/code/,views.LoginSendEmailView.as_view(),nameuser-login-send-email),#登录验证码path(all/user/data/,views.AllUserDataView.as_view(),nameuser-all-user-data),#获取定时任务cache_user_func缓存到cache中的用户数据path(add/result/,views.AddFuncDataView.as_view(),nameuser-add-result),#获取定时任务add_func缓存到cache的计算结果path(task-id/result/,views.UserTaskIdGetDataView.as_view(),nameuser-task-id-data),#通过任务的task-id获取到任务返回值
] 五、调用函数测试
5.1、启动项目
1、启动django项目 python manage.py runserver 2、启动celery异步 #windows系统 celery -A celery_task worker -l info -P eventlet #linux系统 celery -A celery_task worker -l info 3、启动celery定时定时任务也是提交给异步的 celery -A celery_task beat -l info 5.2、测试项目Postman接口工具
1、注册url /user/register/ 2、登录前点击获取验证码返回任务id, url/user/login/code/ 把task_id17ee8389-14ab-4da1-b88a-56446afb4493 复制下来4中可以用到
3、登录code是发送步骤2中发送给邮箱的验证码 ,url/user/login/ 4、通过任务id获取到发送邮箱异步任务的返回值,url/user/task-id/result/
复制2返回值中的task_id,获取该任务的返回值 5、获取定时任务中add_func缓存在cache中的计算数据 6、获取定时任务中cache_user_func缓存到cache中的用户数据 总结
1、异步任务一般是保存文件、发送邮件、耗时操作等
2、定时任务定时处理某些数据。 六、报错
1、先去celery.py中查看定时任务的配置模块的路径是不是有问题千万不要多了一个空格啥的
2、如果通过任务id获取任务的返回值不成功看看是不是添加app.task(ignore_resultTrue),如果是就去掉这个参数配置
