【Python使用】嘿马头条项目从到完整开发教程第9篇：缓存,1 缓存穿透【附代码文档】

本教程的知识点为:简介 1. 内容 2. 目标 产品效果 ToutiaoWeb虚拟机使用说明 数据库 理解ORM 作用 思考： 使用ORM的方式选择 数据库 SQLAlchemy操作 1 新增 2 查询 all() 数据库 分布式ID 1 方案选择 2 头条 使用雪花算法 （代码 toutiao-backend/common/utils/snowflake) 数据库 Redis 1 Redis事务 基本事务指令 Python客户端操作 Git工用流 调试方法 JWT认证方案 JWT & JWS & JWE Json Web Token(JWT) OSS对象存储 存储 需求 方案 使用 缓存 缓存架构 多级缓存 头条项目的方案 缓存数据 缓存 缓存问题 1 缓存 2 缓存 头条项目缓存与存储设计 APScheduler定时任务 定时修正统计数据 RPC RPC简介 1. 什么是RPC RPC 编写客户端 头条首页新闻推荐接口编写 即时通讯 即时通讯简介 即时通讯 Socket.IO 1 简介 优点： 缺点： Elasticsearch 简介与原理 1 简介 属于面向文档的数据库 2 搜索的原理——倒排索引（反向索引）、分析、相关性排序 Elasticsearch 文档 索引文档（保存文档数据） 获取指定文档 判断文档是否存在 单元测试 为什么要测试 测试的分类 什么是单元测试 断言方法的使用：

完整笔记资料代码：https://gitee.com/yinuo112/Backend/tree/master/Python/嘿马头条项目从到完整开发教程/note.md

感兴趣的小伙伴可以自取哦~

全套教程部分目录：

部分文件图片：

缓存

缓存问题

1 缓存穿透

缓存只是为了缓解数据库压力而添加的一层保护层，当从缓存中查询不到我们需要的数据就要去数据库中查询了。如果被黑客利用，频繁去访问缓存中没有的数据，那么缓存就失去了存在的意义，瞬间所有请求的压力都落在了数据库上，这样会导致数据库连接异常。

解决方案：

  1. 约定:对于返回为NULL的依然缓存，对于抛出异常的返回不进行缓存,注意不要把抛异常的也给缓存了。采用这种手段的会增加我们缓存的维护成本，需要在插入缓存的时候删除这个空缓存，当然我们可以通过设置较短的超时时间来解决这个问题。

cache_null

  1. 制定一些规则过滤一些不可能存在的数据，小数据用BitMap，大数据可以用布隆过滤器，比如你的订单ID 明显是在一个范围1-1000，如果不是1-1000之内的数据那其实可以直接给过滤掉。

cache_filter

2 缓存雪崩

缓存雪崩是指缓存不可用或者大量缓存由于超时时间相同在同一时间段失效，大量请求直接访问数据库，数据库压力过大导致系统雪崩。

cache_down

解决方案：

1、给缓存加上一定区间内的随机生效时间，不同的key设置不同的失效时间，避免同一时间集体失效。比如以前是设置10分钟的超时时间，那每个Key都可以随机8-13分钟过期，尽量让不同Key的过期时间不同。

2、采用多级缓存，不同级别缓存设置的超时时间不同，及时某个级别缓存都过期，也有其他级别缓存兜底。

3、利用加锁或者队列方式避免过多请求同时对服务器进行读写操作。

头条项目缓存与存储设计

缓存设计

1 User Cache

用户资料

key类型说明举例
user:{user_id}:profilestringuser_id用户的数据缓存，包括手机号、用户名、头像

用户扩展资料

key类型说明举例
user:{user_id}:profilexstringuser_id用户的性别 生日

用户状态

key类型说明举例
user:{user_id}:statusstringuser_id用户是否可用
key类型说明举例
user:{user_id}:followingzsetuser_id的关注用户[{user_id, update_time}]
key类型说明举例
user:{user_id}:fanszsetuser_id的粉丝用户[{user_id, update_time}]
key类型说明举例
user:{user_id}:artzsetuser_id的文章[{article_id, create_time}]

2 Comment Cache

key类型说明举例
art:{article_id}:commzsetarticle_id文章的评论数据缓存，值为comment_id[{comment_id, create_time}]
comm:{comment_id}:replyzsetcomment_id评论的评论数据缓存，值为comment_id[{'comment_id', create_time}]
comm:{comment_id}string缓存的评论数据

3 Article Cache

key类型说明举例
ch:allstring所有频道
user:{user_id}:chstring用户频道
ch:{channel_id}:art:topzset置顶文章[{article_id, sequence}]
art:{article_id}:infostring文章的基本信息
art:{article_id}:detailstring文章的内容

4 Announcement Cache

key类型说明举例
announcezset[{'json data', announcement_id}]
announce:{announcement_id}string'json data'

持久存储设计

1 阅读历史

key类型说明举例
user:{user_id}:his:readingzset[{article_id, read_time}]

2 搜索历史

key类型说明举例
user:{user_id}:his:searchingzset[{keyword, search_time}]

3 统计数据

key类型说明举例
count:art:readingzset文章阅读数量[{article_id, count}]
count:user:artszset用户发表文章数量[{user_id, count}]
count:art:collectingzset文章收藏数量[{article_id, count}]
count:art:likingzset文章点赞数量[{article_id, count}]
count:art:commzset文章评论数量[{article_id, count}]

头条项目缓存实现

以用户信息数据缓存为例

common/cache/user.py

from flask import current_app
from redis.exceptions import RedisError
import json
from sqlalchemy.orm import load_only

from models.user import User
from . import constants


class UserProfileCache(object):
    """
    用户资料信息缓存
    """

    def __init__(self, user_id):
        self.key = 'user:{}:info'.format(user_id)
        self.user_id = user_id

    def save(self):
        """
        查询数据库保存缓存记录
        :return:
        """
        r = current_app.redis_cluster
        # 查询数据库
        user = User.query.options(load_only(User.name,
                                            User.profile_photo,
                                            User.introduction,
                                            User.certificate)).filter_by(id=self.user_id).first()
        # 判断结果是否存在
        # 保存到redis中
        if user is None:
            try:
                r.setex(self.key, constants.USER_NOT_EXISTS_CACHE_TTL, -1)
            except RedisError as e:
                current_app.logger.error(e)
            return None
        else:
            cache_data = {
                'name': user.name,
                'photo': user.profile_photo,
                'intro': user.introduction,
                'certi': user.certificate
            }
            try:
                r.setex(self.key, constants.UserProfileCacheTTL.get_val(), json.dumps(cache_data))
            except RedisError as e:
                current_app.logger.error(e)
        return cache_data

    def get(self):
        """
        获取用户的缓存数据
        :return:
        """
        r = current_app.redis_cluster

        # 先查询redis
        try:
            ret = r.get(self.key)
        except RedisError as e:
            current_app.logger.error(e)
            ret = None

        if ret is not None:
            # 如果存在记录，读取
            if ret == b'-1':
                # 判断记录值，如果为-1，表示用户不存在
                return None
                # 如果不为-1，需要json转换，返回
            else:
                return json.loads(ret)
        else:
            # 如果记录不存在，
                cache_data = self.save()
                return cache_data

    def clear(self):
        """
        清除用户缓存
        """
        try:
            current_app.redis_cluster.delete(self.key)
        except RedisError as e:
            current_app.logger.error(e)

    def exists(self):
        """
        判断用户是否存在
        """
        # 查询redis
        r = current_app.redis_cluster
        try:
            ret = r.get(self.key)
        except RedisError as e:
            current_app.logger.error(e)
            ret = None

        # 如果缓存记录存在
        if ret is not None:
            if ret == b'-1':
                # 如果缓存记录为-1 ，表示用户不存在
                return False
            else:
                # 如果缓存记录不为-1， 表示用户存在
                return True

        # 如果缓存记录不存在，查询数据库
        else:
            cache_data = self.save()
            if cache_data is not None:
                return True
            else:
                return False

common/cache/constants.py

class BaseCacheTTL(object):
    """
    缓存有效期
    为防止缓存雪崩，在设置缓存有效期时采用设置不同有效期的方案
    通过增加随机值实现
    """
    TTL = 0  # 由子类设置
    MAX_DELTA = 10 * 60  # 随机的增量上限

    @classmethod
    def get_val(cls):
        return cls.TTL + random.randrange(0, cls.MAX_DELTA)


class UserProfileCacheTTL(BaseCacheTTL):
    """
    用户资料数据缓存时间, 秒
    """
    TTL = 30 * 60
接口示例

定义获取当前用户信息的接口

GET /v1_0/user

返回JSON

在toutiao/resources/user/__init__.py中定义路由

user_api.add_resource(profile.CurrentUserResource, '/v1_0/user', endpoint='CurrentUser')

在toutiao/resources/ user/profile.py 中

class CurrentUserResource(Resource):
    """
    用户自己的数据
    """
    method_decorators = [login_required]

    def get(self):
        """
        获取当前用户自己的数据
        """
        user_data = cache_user.UserProfileCache(g.user_id).get()
        user_data['id'] = g.user_id
        return user_data

项目Redis持久存储实现

common/cache/statistic.py

from flask import current_app

from redis.exceptions import ConnectionError


class CountStorageBase(object):
    """
    统计数据存储父类
    """

    key = ''

    @classmethod
    def get(cls, user_id):
        """
        获取指定用户的数值
        :param user_id:
        :return:
        """
        try:
            ret = current_app.redis_master.zscore(cls.key, user_id)
        except ConnectionError as e:
            current_app.logger.error(e)
            ret = current_app.redis_slave.zscore(cls.key, user_id)

        if ret is not None:
            return int(ret)
        else:
            return 0

    @classmethod
    def incr(cls, user_id, increment=1):
        """
        给指定用户机统计数据累计
        :param user_id:
        :param increment:
        :return:
        """
        current_app.redis_master.zincrby(cls.key, user_id, increment)


class UserArticleCountStorage(CountStorageBase):
    """
    用户文章数量存储
    """
    key = 'count:user:arts'


class UserFollowingCountStorage(CountStorageBase):
    """
    用户关注数量
    """
    key = 'count:user:followings'

APScheduler定时任务

APScheduler使用

APScheduler （advanceded python scheduler）是一款Python开发的定时任务工具。

文档地址 [

特点：

  • 不依赖于Linux系统的crontab系统定时，独立运行

  • 可以动态添加新的定时任务，如

下单后30分钟内必须支付，否则取消订单，就可以借助此工具（每下一单就要添加此订单的定时任务）

  • 对添加的定时任务可以做持久保存

1 安装

pip install apscheduler

2 使用方式

from apscheduler.schedulers.background import BackgroundScheduler



# 创建定时任务的调度器对象


scheduler = BackgroundScheduler()



# 定义定时任务


def my_job(param1, param2):
    pass



# 向调度器中添加定时任务


scheduler.add_job(my_job, 'date', args=[100, 'python'])



# 启动定时任务调度器工作


scheduler.start()

3 调度器 Scheduler

负责管理定时任务

  • BlockingScheduler: 作为独立进程时使用
from apscheduler.schedulers.blocking import BlockingScheduler

  scheduler = BlockingScheduler()
  scheduler.start()  # 此处程序会发生阻塞
  • BackgroundScheduler: 在框架程序（如Django、Flask）中使用
from apscheduler.schedulers.background import BackgroundScheduler

  scheduler = BackgroundScheduler()
  scheduler.start()  # 此处程序不会发生阻塞

4 执行器 executors

在定时任务该执行时，以进程或线程方式执行任务

  • ThreadPoolExecutor
from apscheduler.executors.pool import ThreadPoolExecutor
  ThreadPoolExecutor(max_workers)  
  ThreadPoolExecutor(20) # 最多20个线程同时执行

使用方法

executors = {
      'default': ThreadPoolExecutor(20)
  }
  scheduler = BackgroundScheduler(executors=executors)
  • ProcessPoolExecutor
from apscheduler.executors.pool import ProcessPoolExecutor
  ProcessPoolExecutor(max_workers)
  ProcessPoolExecutor(5) # 最多5个进程同时执行

使用方法

executors = {
      'default': ProcessPoolExecutor(3)
  }
  scheduler = BackgroundScheduler(executors=executors)

5 触发器 Trigger

指定定时任务执行的时机

1） date 在特定的时间日期执行
from datetime import date



# 在2019年11月6日00:00:00执行


sched.add_job(my_job, 'date', run_date=date(2009, 11, 6))



# 在2019年11月6日16:30:05


sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5))
sched.add_job(my_job, 'date', run_date='2009-11-06 16:30:05')



# 立即执行


sched.add_job(my_job, 'date')  
sched.start()
2） interval 经过指定的时间间隔执行
  • weeks (int) – number of weeks to wait
  • days (int) – number of days to wait
  • hours (int) – number of hours to wait
  • minutes (int) – number of minutes to wait
  • seconds (int) – number of seconds to wait
  • start_date (datetime|str) – starting point for the interval calculation
  • end_date (datetime|str) – latest possible date/time to trigger on
  • timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations
from datetime import datetime



# 每两小时执行一次


sched.add_job(job_function, 'interval', hours=2)



# 在2010年10月10日09:30:00 到2014年6月15日的时间内，每两小时执行一次


sched.add_job(job_function, 'interval', hours=2, start_date='2010-10-10 09:30:00', end_date='2014-06-15 11:00:00')
3） cron 按指定的周期执行
  • year (int|str) – 4-digit year
  • month (int|str) – month (1-12)
  • day (int|str) – day of the (1-31)
  • week (int|str) – ISO week (1-53)
  • day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)
  • hour (int|str) – hour (0-23)
  • minute (int|str) – minute (0-59)
  • second (int|str) – second (0-59)
  • start_date (datetime|str) – earliest possible date/time to trigger on (inclusive)
  • end_date (datetime|str) – latest possible date/time to trigger on (inclusive)
  • timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations (defaults to scheduler timezone)
# 在6、7、8、11、12月的第三个周五的00:00, 01:00, 02:00和03:00 执行


sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')



# 在2014年5月30日前的周一到周五的5:30执行


sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30')

6 配置方法

方法1
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.executors.pool import ThreadPoolExecutor

executors = {
    'default': ThreadPoolExecutor(20),
}
scheduler = BackgroundScheduler(executors=executors)
方法2
from pytz import utc

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ProcessPoolExecutor

executors = {
    'default': {'type': 'threadpool', 'max_workers': 20},
    'processpool': ProcessPoolExecutor(max_workers=5)
}

scheduler = BackgroundScheduler()



# .. 此处可以编写其他代码





# 使用configure方法进行配置


scheduler.configure(executors=executors)

7 启动

scheduler.start()
  • 对于BlockingScheduler ，程序会阻塞在这，防止退出
  • 对于BackgroundScheduler，程序会立即返回，后台运行

8 扩展

任务管理

方式1

job = scheduler.add_job(myfunc, 'interval', minutes=2)  # 添加任务
job.remove()  # 删除任务
job.pause() # 暂定任务
job.resume()  # 恢复任务

方式2

scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')  # 添加任务    
scheduler.remove_job('my_job_id')  # 删除任务
scheduler.pause_job('my_job_id')  # 暂定任务
scheduler.resume_job('my_job_id')  # 恢复任务
调整任务调度周期
job.modify(max_instances=6, name='Alternate name')

scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')
停止APScheduler运行
scheduler.shutdown()

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/939260.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

谷歌浏览器的扩展程序自动更新设置

谷歌浏览器的扩展程序自动更新设置

谷歌浏览器是全球最受欢迎的网络浏览器之一，其扩展程序更是为用户提供了丰富的功能。然而，随着时间的推移，扩展程序需要更新以修复漏洞、提升性能或增加新功能。本文将详细介绍如何在Chrome中设置扩展程序的自动更新。（本文由http…
阅读更多...
LabVIEW与PLC点位控制及OPC通讯

LabVIEW与PLC点位控制及OPC通讯

在工业自动化中，PLC通过标准协议（如Modbus、Ethernet/IP等）与OPC Server进行数据交换，LabVIEW作为上位机通过OPC客户端读取PLC的数据并进行监控、控制与处理。通过这种方式，LabVIEW能够实现与PLC的实时通信&#xff0c…
阅读更多...
在Windows Server路由和远程访问服务中启用L2TP/IPsec VPN

在Windows Server路由和远程访问服务中启用L2TP/IPsec VPN

背景 路由和远程访问服务（Routing and Remote Access Services，RRAS）是Windows Server上的一个角色，包含很多功能，可以用来搭建VPN。然而，在什么也不做的初始配置中，它只允许PPTP协议连接。然而…
阅读更多...
Android简洁缩放Matrix实现图像马赛克，Kotlin

Android简洁缩放Matrix实现图像马赛克，Kotlin

Android简洁缩放Matrix实现图像马赛克，Kotlin 原理，通过Matrix把一个原图缩小到原先的1/n，然后再把缩小后的小图放大n倍，自然就是马赛克效果（相当于是放大后像素“糊”成一片了）。 import android.content.…
阅读更多...
《Posterior Collapse and Latent Variable Non-identifiability》

《Posterior Collapse and Latent Variable Non-identifiability》

看起来像一篇很有用的paper，而且还是23年的 没看完 后边看不懂了 Abstract 现有的解释通常将后验崩塌归因于由于变分近似而使用神经网络或优化问题。 而本文认为后验崩塌是潜在变量不可识别性的问题(a problem of latent variable non-identifiability) 本文证明了…
阅读更多...
网络视频监控平台/安防监控/视频综合管理Liveweb视频汇聚平台解决方案

网络视频监控平台/安防监控/视频综合管理Liveweb视频汇聚平台解决方案

一、当前现状分析 当前视频资源面临以下问题： 1）不同单位在视频平台建设中以所属领域为单位，设备品牌众多，存在的标准不一，各系统之间也没有统一标准； 2）各单位视频平台建设分散、统筹性差&am…
阅读更多...
【前端爬虫】关于如何获取自己的请求头信息(user-agent和cookie)

【前端爬虫】关于如何获取自己的请求头信息(user-agent和cookie)

注意：由于user-agent和cookie中保存了部分账户信息，所以一定不要随意泄露给他人！！！ 1.首先打开某个页面，点击键盘的F12键进入控制台，或者鼠标右键页面选择打开控制台 2.然后点击控制台上方的网…
阅读更多...
共创共建！葡萄城 SpreadJS 完成 HarmonyOS NEXT 操作系统兼容认证

共创共建！葡萄城 SpreadJS 完成 HarmonyOS NEXT 操作系统兼容认证

最新技术资源（建议收藏） https://www.grapecity.com.cn/resources/ 近日，华为“企业工作必备应用鸿蒙化论坛”在北京圆满落幕，论坛汇聚了众多行业精英和合作伙伴，聚焦讨论企业数字化转型与原生鸿蒙生态融合等话题。葡萄…
阅读更多...
单项链表的学习

单项链表的学习

1:链表概念 链表是⼀种物理存储结构上⾮连续、⾮顺序的存储结构，数据元素的逻辑顺序是通过链表中的指针链接次序实现的。 1：结点 与顺序表不同的是，链表⾥的每节"⻋厢"都是独⽴申请下来的空间，我们称之为“结点 / 结…
阅读更多...
基于大语言模型的多代理下一代制造系统能灵活动态管理制造资源的高效调度方法

基于大语言模型的多代理下一代制造系统能灵活动态管理制造资源的高效调度方法

摘要 论文地址：https://arxiv.org/pdf/2405.16887 随着生产率的提高，客户对多品种、小批量生产的需求也在不断增加，这反过来又对制造系统提出了更高的要求。由于这种需求，当生产任务频繁变化时，传统的制造系统往往无法…
阅读更多...
FPGA-PS端编程1：

FPGA-PS端编程1：

目标 在小梅哥的zynq 7015上，完成以下目标： 读取 S1 按键的电平， 当 S1 按键为按下状态时，驱动 PS LED 以 1S 的频率闪烁(注意理解 1S 的频率闪烁和 1S的时间翻转两种描述之间的差别)， 当 S1 释放后，停止…
阅读更多...
模型 QFD（质量功能展开/质量屋）

模型 QFD（质量功能展开/质量屋）

系列文章 分享 模型，了解更多👉 模型_思维模型目录。将客户需求转化为产品设计。 1 模型 QFD（质量功能展开）的应用 1.1 电信服务及网络维护过程质量改进QFD应用案例 背景介绍： 随着中国加入WTO和国家对电信管制的普遍…
阅读更多...
安装@wangeditor/editor-for-vue失败原因

安装@wangeditor/editor-for-vue失败原因

链接: 安装wangeditor/editor-for-vue失败原因 或者下述命令行: 安装成功可到packa.json里面查看：
阅读更多...
敏捷多模态微型机器人：独特的被动变形轮设计

敏捷多模态微型机器人：独特的被动变形轮设计

大家好！今天来了解一种微型机器人——《An agile multimodal microrobot with architected passively morphing wheels》发表于《SCIENCE ADVANCES》。这个微型机器人，具有独特的设计和卓越的性能。它带有被动变形轮子，这种轮子的设计灵感源自…
阅读更多...
键盘扫描及显示设计实验-微机原理与接口技术课程设计

键盘扫描及显示设计实验-微机原理与接口技术课程设计

1.实验要求 将 8255 单元与键盘及数码管显示单元连接，编写实验程序，扫描键盘输入，并将扫描结果送数码管显示。键盘采用 44 键盘，每个数码管显示值可为 0～F 共 16 个数。实验具体内容如下：将键盘进行编号&am…
阅读更多...
超维机器人油气化工智能巡检解决方案

超维机器人油气化工智能巡检解决方案

随着油气化工行业的快速发展和生产环境的日益复杂，安全、环保和高效的运营管理成为企业发展的关键目标。然而，传统的人工巡检方式面临着许多挑战，包括安全隐患、效率低下和人为误差等问题。为了应对这些挑战，智能化技术的引入为油…
阅读更多...
unity webgl部署到iis报错

unity webgl部署到iis报错

Unable to parse Build/WebGLOut.framework.js.unityweb! The file is corrupt, or compression was misconfigured? (check Content-Encoding HTTP Response Header on web server) iis报错的 .unityweb application/octet-stream iis中添加 MIME类型 .data applicatio…
阅读更多...
【Lua热更新】下篇 -- 更新中

【Lua热更新】下篇 -- 更新中

上篇链接：【Lua热更新】上篇 文章目录 三、xLua热更新📖1.概述📚︎2.导入xLua框架🔖3. C#调用Lua3.1Lua解析器3.2Lua文件夹的重定向3.3Lua解析器管理器3.4全局变量获取3.5全局函数获取3.6映射到List和Dictionary3.7映射到类3.8映…
阅读更多...
基于 uniapp 开发 android 播放 webrtc 流

基于 uniapp 开发 android 播放 webrtc 流

一、播放rtsp协议流 如果 webrtc 流以 rtsp 协议返回&#xff0c;流地址如&#xff1a;rtsp://127.0.0.1:5115/session.mpg&#xff0c;uniapp的 <video> 编译到android上直接就能播放&#xff0c;但通常会有2-3秒的延迟。 二、播放webrtc协议流 如果 webrtc 流以 webrt…
阅读更多...
Redis应用—7.大Value处理方案

Redis应用—7.大Value处理方案

大纲 1.⽅案设计 2.安装与配置环境 1.⽅案设计 步骤一&#xff1a;首先需要配置一个crontab定时调度shell脚本&#xff0c;然后该脚本每天凌晨会通过rdbtools⼯具解析Redis的RDB⽂件&#xff0c;接着对解析出的内容进行过滤&#xff0c;把RDB⽂件中的⼤key导出到CSV⽂件。 步…
阅读更多...
最新文章

Copyright @ 2022~2023