🏆🏆🏆教程全知识点简介：1.APScheduler任务调度涵盖安装配置、使用方式、调度器Scheduler、执行器executors、触发器Trigger等核心组件。2. RPC远程过程调用包括RPC概念、背景用途、优缺点分析。3. Protocol Buffers数据序列化涉及文档结构、注释语法、数据类型、枚举类型、消息类型（字段编号、字段规则、嵌套类型、保留字段、默认值）。4. 客户端开发包含头条首页新闻推荐接口编写。5. 即时通讯技术涵盖需求场景、传统推送实现、Socket.IO（Python服务器端开发、事件处理）。6. Elasticsearch搜索引擎包括简介原理、倒排索引、分析器、相关性排序、集群概念、IK中文分析器、索引类型、文档操作（索引文档、获取文档、判断存在、更新删除）、Logstash数据导入、查询（基本查询、高级查询）、全文检索实现、Python客户端使用、联想提示（拼写纠错、自动补全）。7. 单元测试涵盖测试分类、基本写法、测试必要性。8. 服务器部署包括Gunicorn、Supervisor配置管理。9. 项目开发流程涉及产品介绍、原型图UI图、技术架构、开发环境（ToutiaoWeb虚拟机、Pycharm远程开发）。10. 数据库技术包含ORM理解、SQLAlchemy映射构建、数据库连接设置、模型类字段选项。11. 分布式系统涵盖分布式ID方案选择、Twitter Snowflake算法（64位ID划分、最大取值计算、移位偏移计算、序号循环掩码、时间戳处理）。12. Redis数据库包括Redis持久化机制。13. Git工作流涵盖Gitflow工作流（工作方式、历史分支、功能分支、发布分支、维护分支）、调试方法。14. 身份认证技术包含JWT、JWS、JWE概念、Python库使用、项目封装实施方案。15. 对象存储涉及OSS对象存储、七牛云存储服务。16. 缓存系统包括缓存架构、缓存数据保存方式、缓存有效期TTL、缓存淘汰策略、缓存问题（缓存穿透、缓存雪崩）、头条项目缓存设计（User Cache、Article Cache、Announcement Cache）、持久存储设计（阅读历史、搜索历史、统计数据）。

📚📚仓库code.zip 👉直接-->：   https://gitee.com/yinuo112/Backend/blob/master/Python/嘿马头条项目从到完整开发教程/note.md    🍅🍅

✨ 本教程项目亮点

🧠 知识体系完整：覆盖从基础原理、核心方法到高阶应用的全流程内容
💻 全技术链覆盖：完整前后端技术栈，涵盖开发必备技能
🚀 从零到实战：适合 0 基础入门到提升，循序渐进掌握核心能力
📚 丰富文档与代码示例：涵盖多种场景，可运行、可复用
🛠 工作与学习双参考：不仅适合系统化学习，更可作为日常开发中的查阅手册
🧩 模块化知识结构：按知识点分章节，便于快速定位和复习
📈 长期可用的技术积累：不止一次学习，而是能伴随工作与项目长期参考

🎯🎯🎯全教程总章节

🚀🚀🚀本篇主要内容

缓存

缓存问题

1 缓存穿透

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

解决方案：

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

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

2 缓存雪崩

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

解决方案：

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

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

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

头条项目缓存与存储设计

缓存设计

1 User Cache

用户资料

用户扩展资料

用户状态

2 Comment Cache

3 Article Cache

4 Announcement Cache

持久存储设计

1 阅读历史

2 搜索历史

3 统计数据

头条项目缓存实现

以用户信息数据缓存为例

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

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

@classmethod
def get_val(cls):
    return cls.TTL + r