最新要闻

广告

手机

iphone11大小尺寸是多少？苹果iPhone11和iPhone13的区别是什么？

iphone11大小尺寸是多少？苹果iPhone11和iPhone13的区别是什么？

 警方通报辅警执法直播中被撞飞：犯罪嫌疑人已投案

警方通报辅警执法直播中被撞飞：犯罪嫌疑人已投案

家电

  1. 首页
  2. 移动通信
  3. 内容

今日热议：【django-vue】celery延迟任务、定时任务 django中使用celery 秒杀功能 双写一致性 首页轮播图定时更新 课程前端页面

来源：博客园
目录
  • 上节回顾
    • 字符编码
    • django-redis
  • 今日内容
    • 1 celery 执行异步任务，延迟任务，定时任务
      • 延时任务
      • 定时任务
    • 2 django中使用celery
      • 2.1 秒杀功能
        • 2.1.1 视图
        • 2.1.2 任务 order_task.py
        • 2.1.3 前端Sckill.vue
      • 2.2 celery中使用django orm
    • 3 轮播图接口加缓存
    • 4 双写一致性
    • 4 首页轮播图定时更新
    • 5 课程前端页面
    • 6 课程功能表分析
  • 练习

上节回顾

# 1 redis 是什么---》非关系型数据库，数据存在内存中，速度快，5大数据类型，主要用来做缓存# 2 安装  图形化客户端# 3 python 操作redis-两种连接方式    -连接池的链接， 池对象是单例的    # 4 字符串，列表，hash     -----集合   有序集合# 5 字符串# 6 hash操作---》key - vuale这种形式---》底层基于数组 ,根据key值使用hash函数运算，得到一个地址，把value值存到地址中# 7 列表# 8 通用操作# 9 管道 pipline---》多条命令先放到管道中不执行 ，执行execute，一次性执行管道中的所有命令事务    # 10 django中使用-自己写     -写个包 POOL=拿到连接池        -以后在用的地方，导入POOL，从池中拿一个连接用即可 conn            -django-redis    -配置文件配置    -django缓存  cache.set  cache.set        -导入conn=get_redis_connection()        conn.set("name","lqz")    # 11 celery  分布式异步任务框架      -框架，能够做 异步任务，定时任务，延迟任务     -独立的，架构         -消息中间件：存储任务的地方,借助于第三方 redis        -任务执行单元：worker 真正执行任务的进程，执行消息中间件中的任务(函数)        -结果存储：任务执行的的返回值，放到里面  redis         -快速使用        app=Celery(参数)        @app.task        def add():            pass       add.delay()        worker执行        把结果存到redis        查看结果                     -包结构    -clelery_task        -__init__        -celery.py    include            -task1.py            -task2.py         -提交任务     -查看任务执行结果           # 补充 字符编码重点梳理1 ascii  英文字母大小写，数字，标点符号         -127 够了   1个字节  8个比特位[每个比特位只能放0或1]  2的8次方中变化 256种变化        -ab!  00001001  000010010 000010010            2 中国，表示中文       GBK编码：用两个字节 表示一个字符            2的16次方:65536                3 韩国，表示韩文    998      文        998      한                    4 unicode编码  把世界上所有的象形文字都有个数字对应   4个字节表示一个字符    998      文        28893    한            5 unicode编码存储---》硬盘上---》ab文한   4个字节 4个字节 4个字节 4个字节    -utf-8  unicode编码 编码，可变长        a   一个字节        b   一个字节        文   两个字节        한   三个字节                1个字节  1个字节   两个字节  三个字节        0 真正的存储的值        00 真正的值  00 真正的值        000

字符编码

# 补充 字符编码重点梳理1 ascii  英文字母大小写，数字，标点符号         -127 够了   1个字节  8个比特位[每个比特位只能放0或1]  2的8次方中变化 256种变化        -ab!  00001001  000010010 000010010            2 中国，表示中文       GBK编码：用两个字节 表示一个字符            2的16次方:65536                3 韩国，表示韩文    998      文        998      한                    4 unicode编码  把世界上所有的象形文字都有个数字对应   4个字节表示一个字符    998      文        28893    한            5 unicode编码存储---》硬盘上---》ab文한   4个字节 4个字节 4个字节 4个字节    -utf-8  unicode编码 编码，可变长        a   一个字节        b   一个字节        文   两个字节        한   三个字节                1个字节  1个字节   两个字节  三个字节        0 真正的存储的值        00 真正的值  00 真正的值        000         # 补充        文件分为：文本文件、其他文件。文本文件涉及到字符编码。ascii 英文字母大小写，数字，标点符号电脑   <--ascii--  英文字母ascii表存储字母和二进制的对应关系。乱码：不同的编码，二进制和字符的对应关系不一样。不同的二进制在不同的编码中，和字符对应关系不一样。二进制 --gbk--> 汉字二进制 --ascii--> 中文会乱码 通过字节开头0的个数，来判断取几个字节。

django-redis

# 10 django中使用-自己写:     -写个包 POOL=拿到连接池        -以后在用的地方，导入POOL，从池中拿一个连接用即可 conn                -django-redis：    以下两个方案都使用相同的配置：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",        }        }}        方案一：使用django的缓存 缓存的数据存入redis    -1.配置文件配置          1.如果配置了，就会存储到redis里。            2.如果没配置，就会存储到内存中，项目重启会导致缓存消失。    -2.使用 cache.get  cache.set 存取数据        特点：            1. 在redis会专门建立一个文件夹                2. cache.get()里面写的键，和最终存储到redis的键是不一样的                3. 可以存储任意数据类型，底层基于pickle模块            方案二：     -1.配置文件配置         -2.导入conn=get_redis_connection()        conn.set("name","lqz")

使用方案一，存储到redis，会建立一个文件夹，可以存对象（二进制形式）：


(资料图片仅供参考)

今日内容

1 celery 执行异步任务，延迟任务，定时任务

# 1 异步任务   任务.delay(参数)     # 2 延迟任务任务.apply_async(args=[参数],eta=时间对象)         """注意：如果没有修改时区，需要使用utc事件"""    # 3 定时任务-需要启动beat和启动worker    -beat    定时提交任务的进程---》配置在app.conf.beat_schedule的任务        -worker  执行任务的            ### 使用步骤    # 第一步：在celery的py文件中写入        app.conf.timezone = "Asia/Shanghai"        # 是否使用UTC        app.conf.enable_utc = False        # celery的配置文件#####        # 任务的定时配置        app.conf.beat_schedule = {            "send_sms": {                "task": "celery_task.user_task.send_sms",                # "schedule": timedelta(seconds=3),  # 时间对象                # "schedule": crontab(hour=8, day_of_week=1),  # 每周一早八点                "schedule": crontab(hour=9, minute=43),  # 每天9点43                "args": ("18888888", "6666"),            },        }                ## 第二步：启动beat        celery -A celery_task beat -l info                ## 第三步：启动worker        celery -A celery_task worker -l info -P eventlet                # 注意点：1 启动命令的执行位置，如果是包结构，一定在包这一层    2 include=["celery_task.order_task"],路径从包名下开始导入，因为我们在包这层执行的命令

延时任务

获取5s后的时间对象：（apply_async）

celery默认使用utc时间。时间计算需要使用timedelta对象。

查看当地时间（中国：东八区）和utc时间的区别：

定时任务

定时任务需要在app的配置项中配置（也就是app.conf这个变量）：

配置好这个时区之后，就不需要使用utc时间，直接使用本地时间即可。

定时任务配置项：

配置了一个定时任务叫send_sms_task。每隔5s发送一条短信。

配置另外一个任务：

day_of_week=1，表示每周一。

启动beat：原来任务是我们手动提交的，现在任务是beat进程提交的。

beat根据我们的定时任务，每隔5s提交一次任务。worker里面有任务，就会去执行。

2 django中使用celery

# 补充：如果在公司中，只是做定时任务，还有一个更简单的框架APSchedule：https://blog.csdn.net/qq_41341757/article/details/118759836                # 使用步骤：-1 把咱们写的包，复制到项目目录下    -luffy_api        -celery_task #celery的包路径            -luffy_api  #源代码路径                -2 在使用提交异步任务的位置，导入使用即可    -视图函数中使用，导入任务        -任务.delay()  # 提交任务                    -3 启动worker，如果有定时任务，启动beat        -4 等待任务被worker执行        -5 在视图函数中，查询任务执行的结果         """使用celery提高了项目的并发量"""

2.1 秒杀功能

# 秒杀逻辑分析1 前端秒杀按钮，用户点击---》发送ajax请求到后端    2 视图函数---》提交秒杀任务---》借助于celery，提交到消息中间件中了    3 当次秒杀的请求，就回去了，携带者任务id号在前端    4 前端开启定时任务，每隔3s钟，带着任务，向后端发送请求，查看是否秒杀成功    5 后端的情况    1 任务还在等待被执行----》返回给前端，前端继续每隔3s发送一次请求        2 任务执行完了，秒杀成功了---》返回给前端，恭喜您秒杀成功--》关闭前端定时器        3 任务执行完了，秒杀失败了---》返回给前端，秒杀失败--》关闭前端定时器

2.1.1 视图

#### 秒杀逻辑，CBVfrom rest_framework.viewsets import ViewSetfrom celery_task.order_task import sckill_taskfrom celery_task.celery import appfrom celery.result import AsyncResultclass SckillView(ViewSet):    @action(methods=["GET"], detail=False)    def sckill(self, request):        a = request.query_params.get("id")        # 使用异步，提交一个秒杀任务        res = sckill_task.delay(a)        return APIResponse(task_id=res.id)    @action(methods=["GET"], detail=False)    def get_result(self, request):        task_id = request.query_params.get("task_id")        a = AsyncResult(id=task_id, app=app)        if a.successful():            result = a.get()            if result:                return APIResponse(msg="秒杀成功")            else:                return APIResponse(code=101, msg="秒杀失败")        elif a.status == "PENDING":            print("任务等待中被执行")            return APIResponse(code=666, msg="还在秒杀中")

2.1.2 任务 order_task.py

# 秒杀任务import randomimport time@app.taskdef sckill_task(good_id):    # 生成订单，减库存，都要在一个事务中    print("商品%s：秒杀开始" % good_id)    # 这个过程，可能是1,2,3s中的任意一个    time.sleep(random.choice([6, 7, 9]))    print("商品%s秒杀结束" % good_id)    return random.choice([True, False])

2.1.3 前端Sckill.vue

<script>import Header from "@/components/Header";import Banner from "@/components/Banner";import Footer from "@/components/Footer";export default {  name: "Sckill",  data() {    return {      task_id: "",      t: null    }  },  methods: {    handleSckill() {      this.$axios.get(this.$settings.BASE_URL + "/user/sckill/sckill/?id=999").then(res => {        this.task_id = res.data.task_id        this.t = setInterval(() => {          this.$axios.get(this.$settings.BASE_URL + "/user/sckill/get_result/?task_id=" + this.task_id).then(res => {            if (res.data.code == 666) {              //如果秒杀任务还没执行，定时任务继续执行              console.log(res.data.msg)            } else {              // 秒杀结束，无论成功失败，这个定时任务都结束              clearInterval(this.t)              this.t = null              this.$message(res.data.msg)            }          })        }, 2000)      }).catch(res => {      })    }  }}</script>

2.2 celery中使用django orm

-1 把咱们写的包，复制到项目目录下    -luffy_api        -celery_task #celery的包路径            celery.py  # 一定不要忘了一句话                import os                os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffy_api.settings.dev")            -luffy_api  #源代码路径                -2 在使用提交异步任务的位置，导入使用即可    -视图函数中使用，导入任务        -任务.delay()  # 提交任务                    -3 启动worker，如果有定时任务，启动beat        -4 等待任务被worker执行        -5 在视图函数中，查询任务执行的结果             # 重点：celery中使用djagno，有时候，任务中会使用django的orm，缓存，表模型。。。。一定要加os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffy_api.settings.dev")

写一个视图函数，发送短信：

启动worker\beat\django:

在在celery任务函数中集成django（比如django orm），需要在包内加入以下代码：

在脚本中运行django。

实例：

这样相当于把django的环境变量，也添加到了celery包内。

3 轮播图接口加缓存

# 1 网站首页被访问的频率很高，瞬间 1w个人在访问，首页的轮播图接口会执行1w次，1w次查询轮播图标的sql在执行，轮播图基本不变# 2 想一种方式，让这1w个访问，效率更高一些，不查数据库了，直接走缓存--》redis--》效率高# 3 现在的逻辑变成了1 轮播图接口请求来了，先去缓存中看，如果有，直接返回    2 如果没有，查数据库，然后把轮播图数据，放到redis中，缓存起来        # 改接口    # 加入缓存的轮播图接口    def list(self, request, *args, **kwargs):        # 查看缓存有没有数据，如果没有，再走数据库        banner_list = cache.get("banner_list")        if banner_list:            print("走了缓存")            return APIResponse(data=banner_list)        else:  # 走数据库            print("走了数据库")            res = super().list(request, *args, **kwargs)            # 把序列化后的数据存到缓存中，redis中            cache.set("banner_list", res.data)            return APIResponse(data=res.data) # 补充查询所有的接口适合加缓存。mysql查询速度慢的接口的适合加缓存。

4 双写一致性

# 加入缓存后，缓存中有数据，先去缓存拿，但是如果mysql中数据变了，缓存不会自动变化，出现数据不一致的问题mysql 和 缓存数据库 数据不一致        # 双写一致性写入mysql，redis没动，数据不一致存在问题    数据库数据修改了、但是缓存的数据没变。        # 如何解决1 修改数据，删除缓存    2 修改数据，更新缓存    3 定时更新缓存  ---> 实时性差    不同的方法，解决不同的需求。        # 定时任务：celery

4 首页轮播图定时更新

实现逻辑：

# 启动django# 启动worker# 启动beat# 第一次访问：查的数据库，放入了缓存# 以后再访问，走缓存# 一旦mysql数据改了，缓存可能不一致# 当时我们定时更新，最终保持了一致

步骤：

# 第一步：在celery配置定时任务app.conf.beat_schedule = {    "update_banner": {        "task": "celery_task.home_task.update_banner",        "schedule": timedelta(seconds=3),  # 时间对象    },}# 第二步：启动worker，启动beat# update_banner任务的代码from home.models import Bannerfrom home.serializer import BannerSerializerfrom django.core.cache import cachefrom django.conf import settings@app.taskdef update_banner():    # 只要这个任务一执行，就更新轮播图的缓存    banners = Banner.objects.all().filter(is_delete=False, is_show=True).order_by("orders")    ser = BannerSerializer(instance=banners, many=True)    for item in ser.data:        item["image"] = settings.BACKEND_URL + item["image"]    cache.set("banner_list", ser.data)  # 会出问题，轮播图地址显示不全    return True

视图类有request对象，不用拼地址，在celery的task函数里需要自己拼接地址：

没有深拷贝，所以会直接改变列表：

5 课程前端页面

# 免费课，实战课，轻课-不要钱的，免费的    -实战课：python面向对象  39.9     -线下课程完整版的视频   19999                # 课程列表页数据1 新建3个页面组件    FreeCourserView        ActualCourserView        LightCourseView    2 配置路由      {        path: "/free-course",        name: "free",        component: FreeCourserView    },    {        path: "/actual-course",        name: "actual-course",        component: ActualCourserView    },    {        path: "/light-course",        name: "light-course",        component: LightCourseView    },3 复制代码    
<script>import Header from "@/components/Header"import Footer from "@/components/Footer"export default {  name: "Course",  data() {    return {      category: 0,    }  },  components: {    Header,    Footer,  }}</script>

6 课程功能表分析

# 轻课 实战课 免费课    需要几个表-所有课程使用一个表  通过类型区分，但是可能出现字段不一样，数据量越来越多，导致表查询速度慢    -一种课程一个表        # 实战课线---》分析出5张表-课程分类表  一个分类下有多个课程跟课程一对多-实战课表  一个实战课，会有多个章节，跟章节一对多    -章节表   一个章节下有多个课时，章节和课时一对多    -课时表    -老师表   跟实战课一对多            -正常课程 有评论，需要建立评论表，评论板块没写，这个表就不写了
from django.db import modelsfrom utils.common_model import BaseModel# Create your models here.# 5张：课程分类，实战课，章节，课时，老师表# 课程分类表class CourseCategory(BaseModel):    """分类"""    name = models.CharField(max_length=64, unique=True, verbose_name="分类名称")    class Meta:        db_table = "luffy_course_category"        verbose_name = "分类"        verbose_name_plural = verbose_name    def __str__(self):        return "%s" % self.name# 实战课表class Course(BaseModel):    """课程"""    course_type = (        (0, "付费"),        (1, "VIP专享"),        (2, "学位课程")    )    level_choices = (        (0, "初级"),        (1, "中级"),        (2, "高级"),    )    status_choices = (        (0, "上线"),        (1, "下线"),        (2, "预上线"),    )    name = models.CharField(max_length=128, verbose_name="课程名称")    course_img = models.ImageField(upload_to="courses", max_length=255, verbose_name="封面图片", blank=True, null=True)    course_type = models.SmallIntegerField(choices=course_type, default=0, verbose_name="付费类型")    brief = models.TextField(max_length=2048, verbose_name="详情介绍", null=True, blank=True)    level = models.SmallIntegerField(choices=level_choices, default=0, verbose_name="难度等级")    pub_date = models.DateField(verbose_name="发布日期", auto_now_add=True)    period = models.IntegerField(verbose_name="建议学习周期(day)", default=7)    attachment_path = models.FileField(upload_to="attachment", max_length=128, verbose_name="课件路径", blank=True,null=True)    status = models.SmallIntegerField(choices=status_choices, default=0, verbose_name="课程状态")    students = models.IntegerField(verbose_name="学习人数", default=0) # 跟用户表做关联，优化字段    sections = models.IntegerField(verbose_name="总课时数量", default=0)    pub_sections = models.IntegerField(verbose_name="课时更新数量", default=0)    price = models.DecimalField(max_digits=6, decimal_places=2, verbose_name="课程原价", default=0)    # on_delete 级联删除，出了级联删除外，还有 很多其他的，暂时先不讲    teacher = models.ForeignKey("Teacher", on_delete=models.DO_NOTHING, null=True, blank=True, verbose_name="授课老师")    course_category = models.ForeignKey("CourseCategory", on_delete=models.SET_NULL, db_constraint=False, null=True,blank=True, verbose_name="课程分类")    class Meta:        db_table = "luffy_course"        verbose_name = "课程"        verbose_name_plural = "课程"    def __str__(self):        return "%s" % self.name# 章节表class CourseChapter(BaseModel):    """章节"""    # 章节跟课程一对多，关联字段写在多的一方    # related_name 暂时先不讲    course = models.ForeignKey("Course", related_name="coursechapters", on_delete=models.CASCADE, verbose_name="课程名称")    chapter = models.SmallIntegerField(verbose_name="第几章", default=1)    name = models.CharField(max_length=128, verbose_name="章节标题")    summary = models.TextField(verbose_name="章节介绍", blank=True, null=True)    pub_date = models.DateField(verbose_name="发布日期", auto_now_add=True)    class Meta:        db_table = "luffy_course_chapter"        verbose_name = "章节"        verbose_name_plural = verbose_name    def __str__(self):        return "%s:(第%s章)%s" % (self.course, self.chapter, self.name)# 课时表class CourseSection(BaseModel):    """课时"""    section_type_choices = (        (0, "文档"),        (1, "练习"),        (2, "视频")    )    # 课时跟章节一对多，关联写段写在多的一方    chapter = models.ForeignKey("CourseChapter", related_name="coursesections", on_delete=models.CASCADE,verbose_name="课程章节")    name = models.CharField(max_length=128, verbose_name="课时标题")    orders = models.PositiveSmallIntegerField(verbose_name="课时排序")    section_type = models.SmallIntegerField(default=2, choices=section_type_choices, verbose_name="课时种类")    section_link = models.CharField(max_length=255, blank=True, null=True, verbose_name="课时链接",help_text="若是video，填vid,若是文档，填link")    duration = models.CharField(verbose_name="视频时长", blank=True, null=True, max_length=32)  # 仅在前端展示使用    pub_date = models.DateTimeField(verbose_name="发布时间", auto_now_add=True)    free_trail = models.BooleanField(verbose_name="是否可试看", default=False)    class Meta:        db_table = "luffy_course_section"        verbose_name = "课时"        verbose_name_plural = verbose_name    def __str__(self):        return "%s-%s" % (self.chapter, self.name)# 老师表class Teacher(BaseModel):    """导师"""    role_choices = (        (0, "讲师"),        (1, "导师"),        (2, "班主任"),    )    name = models.CharField(max_length=32, verbose_name="导师名")    role = models.SmallIntegerField(choices=role_choices, default=0, verbose_name="导师身份")    title = models.CharField(max_length=64, verbose_name="职位、职称")    signature = models.CharField(max_length=255, verbose_name="导师签名", help_text="导师签名", blank=True, null=True)    image = models.ImageField(upload_to="teacher", null=True, verbose_name="导师封面")    brief = models.TextField(max_length=1024, verbose_name="导师描述")    class Meta:        db_table = "luffy_teacher"        verbose_name = "导师"        verbose_name_plural = verbose_name    def __str__(self):        return "%s" % self.name                # 执行迁移命令

练习

# 秒杀功能，自己写完 ok# 轮播图加缓存 ok -高级一些，写一个可以缓存的视图类，以后只要继承这个视图类，接口就有缓存，不继承就没有缓存       # 轮播图定时更新 ok# 课程5个表迁移完成 ok

关键词：