python3 自动更新的缓存类

这个类会在后台自动更新缓存数据,你只需要调用方法来获取数据即可。


自动更新缓存类

以下是 AutoUpdatingCache 类的实现:

import threading
import timeclass AutoUpdatingCache:def __init__(self, update_function, expiry_time=60):"""初始化缓存类。:param update_function: 一个函数,用于生成或更新缓存数据。:param expiry_time: 缓存的更新周期(秒)。"""self.update_function = update_functionself.expiry_time = expiry_timeself.cache_data = Noneself.last_updated = 0self.lock = threading.Lock()self._start_background_update()def _start_background_update(self):# 启动后台线程更新缓存self.update_thread = threading.Thread(target=self._update_cache_periodically)self.update_thread.daemon = Trueself.update_thread.start()def _update_cache_periodically(self):while True:current_time = time.time()if current_time - self.last_updated >= self.expiry_time:self._update_cache()time.sleep(1)  # 每秒检查一次def _update_cache(self):with self.lock:try:print("Updating cache...")new_data = self.update_function()self.cache_data = new_dataself.last_updated = time.time()print("Cache updated!")except Exception as e:print(f"Error updating cache: {e}")def get_data(self):with self.lock:if self.cache_data is not None:return self.cache_dataelse:return "Cache is initializing, please try again later."

使用说明

  1. 定义一个数据生成函数

    首先,需要定义一个用于生成或更新缓存数据的函数。这个函数可以是任何耗时的操作,例如从数据库查询、计算复杂结果等。

    import timedef generate_cache_data():# 模拟耗时操作time.sleep(5)return {"value": "fresh data", "timestamp": time.time()}
    
  2. 创建缓存类的实例

    将数据生成函数传递给 AutoUpdatingCache 类,并设置缓存更新周期。

    cache = AutoUpdatingCache(update_function=generate_cache_data, expiry_time=30)
    
  3. 获取缓存数据

    在需要的地方调用 get_data() 方法即可获取缓存数据。

    data = cache.get_data()
    print(data)
    

完整示例

将以上步骤组合起来:

import threading
import timeclass AutoUpdatingCache:def __init__(self, update_function, expiry_time=60):self.update_function = update_functionself.expiry_time = expiry_timeself.cache_data = Noneself.last_updated = 0self.lock = threading.Lock()self._start_background_update()def _start_background_update(self):self.update_thread = threading.Thread(target=self._update_cache_periodically)self.update_thread.daemon = Trueself.update_thread.start()def _update_cache_periodically(self):while True:current_time = time.time()if current_time - self.last_updated >= self.expiry_time:self._update_cache()time.sleep(1)def _update_cache(self):with self.lock:try:print("Updating cache...")new_data = self.update_function()self.cache_data = new_dataself.last_updated = time.time()print("Cache updated!")except Exception as e:print(f"Error updating cache: {e}")def get_data(self):with self.lock:if self.cache_data is not None:return self.cache_dataelse:return "Cache is initializing, please try again later."# 数据生成函数
def generate_cache_data():time.sleep(5)  # 模拟耗时操作return {"value": "fresh data", "timestamp": time.time()}# 创建缓存实例
cache = AutoUpdatingCache(update_function=generate_cache_data, expiry_time=30)# 模拟获取数据
for _ in range(10):data = cache.get_data()print(data)time.sleep(10)

代码解释

  • AutoUpdatingCache 类

    • init 方法:
      • 初始化缓存,设置数据生成函数和缓存更新周期。
      • 启动后台线程 _update_cache_periodically
    • _update_cache_periodically 方法:
      • 无限循环,每隔一秒检查缓存是否需要更新。
      • 如果当前时间距离上次更新时间超过了 expiry_time,则调用 _update_cache
    • _update_cache 方法:
      • 使用 update_function 更新缓存数据。
      • 使用锁机制 threading.Lock 确保线程安全。
    • get_data 方法:
      • 获取缓存数据。
      • 如果缓存数据为空(初始化中),返回提示信息。
  • 数据生成函数

    • generate_cache_data 函数模拟一个耗时操作,生成新的缓存数据。
  • 使用示例

    • 创建缓存实例并在循环中每隔 10 秒获取一次数据,观察缓存的更新情况。

注意事项

  • 线程安全:

    • 使用 threading.Lock 确保在多线程环境下数据访问的安全性。
  • 异常处理:

    • 在更新缓存时,捕获可能的异常,防止线程崩溃。
  • 后台线程:

    • 将线程设置为守护线程(daemon=True),使得主程序退出时,线程自动结束。

应用场景

你可以将这个缓存类应用在 Web 应用程序中,例如在 Sanic 的路由中:

from sanic import Sanic
from sanic.response import jsonapp = Sanic("CacheApp")@app.route("/data")
async def get_cached_data(request):data = cache.get_data()return json({"data": data})if __name__ == "__main__":# 确保缓存在应用启动前初始化cache = AutoUpdatingCache(update_function=generate_cache_data, expiry_time=30)app.run(host="0.0.0.0", port=8000)

这样,用户在访问 /data 路由时,总是能得到缓存中的数据,而缓存会在后台自动更新,不会因为更新缓存而导致请求超时。


😊

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

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【微服务】Nacos

一、安装 1、官网地址:https://nacos.io/download/nacos-server/ 2、启动:找到bin目录下的startup.cmd双击启动,或者打开一个命令窗口输入: startup.cmd -m standalone双击启动后如下:可以访问控制台地址 访问后的…

学习笔记032——Spring学习笔记

文章目录 一、Spring开发步骤二、Spring配置文件1、Bean标签基本配置2、Bean标签范围配置3、Bean生命周期配置4、Bean实例化三种方式5、Bean的依赖注入概念6、Bean的依赖注入方式【第一种:set方法注入】【第二种:构造方法注入】 7、Bean的依赖注入的数据…

某科技研发公司培训开发体系设计项目成功案例纪实

某科技研发公司培训开发体系设计项目成功案例纪实 ——建立分层分类的培训体系,加强培训跟踪考核,促进培训成果实现 【客户行业】科技研发行业 【问题类型】培训开发体系 【客户背景】 某智能科技研发公司是一家专注于智能科技、计算机软件技术开发与…

Elasticsearch:Retrievers 介绍

检索器(retrievers)是 Elasticsearch 中搜索 API 中添加的新抽象层。它们提供了在单个 _search API 调用中配置多阶段检索管道的便利。此架构通过消除对复杂搜索查询的多个 Elasticsearch API 调用的需求,简化了应用程序中的搜索逻辑。它还减…

Python学习34天

import random class Game: peo0 rob0 # # def __init__(self,peo,rob): # self.peopeo # self.robrob def Play(self): """ 石头剪刀布游戏,0代表石头,1代见到,2代表石头 …

hive的存储格式

1) 四种存储格式 hive的存储格式分为两大类:一类纯文本文件,一类是二进制文件存储。 Hive支持的存储数据的格式主要有:TEXTFILE、SEQUENCEFILE、ORC、PARQUET 第一类:纯文本文件存储 textfile: 纯文本文件存储格式…

solr 远程命令执行 (CVE-2019-17558)

目录 漏洞描述 执行漏洞py脚本,取得shell连接 EXP 漏洞描述 Apache Velocity是一个基于Java的模板引擎,它提供了一个模板语言去引用由Java代码定义的对象。Velocity是Apache基金会旗下的一个开源软件项目,旨在确保Web应用程序在表示层和业…

uname -m(machine) 命令用于显示当前系统的机器硬件架构(Unix Name)

文章目录 关于 arm64 架构检查是否安装了 Rosetta 2其他相关信息解释:命令功能:示例: dgqdgqdeMac-mini / % uname -m arm64您运行的 uname -m 命令显示您的系统架构是 arm64。这意味着您的 Mac Mini 使用的是 Apple 的 M1 或更新的芯片&…

实现在两台宿主机下的docker container 中实现多机器通讯

基于我的实验背景 上位机:ubuntu 20.04 (docker humble 22.04) 下位机:ubuntu 22.04(docker noetic 20.04) 目标:实现在上位机中的docker container 容器的22.04环境去成功远程访问 非同网段的下位机的20.04的contai…

(计算机组成原理)期末复习

第一章 计算机的基本组成:硬件软件(程序)计算机系统 软件有系统软件(系统管理工具),应用软件 计算机硬件:包括主机和外设,主机包括CPU和内存,***CPU由运算器和控制器所组…

QML TableView 实例演示 + 可能遇到的一些问题(Qt_6_5_3)

一、可能遇到的一些问题 Q1:如何禁用拖动? 在TableView下加一句代码即可: interactive: false 补充:这个属性并不专属于TableView,而是一个通用属性。很多Controls下的控件都可以使用,其主要作用就是控…

【vue3实现微信小程序】从轮播图到公告栏的前端开发之旅

快速跳转: 我的个人博客主页👉:Reuuse博客 新开专栏👉:Vue3专栏 参考文献👉:uniapp官网 免费图标👉:阿里巴巴矢量图标库 ❀ 感谢支持!☀ 前情提要 &#x…

【大数据学习 | Spark-Core】RDD的缓存(cache and checkpoint)

1. 单应用缓存:cache 1.1 cache算子 cache算子能够缓存中间结果数据到各个executor中,后续的任务如果需要这部分数据就可以直接使用避免大量的重复执行和运算。 rdd 存储级别中默认使用的算子cache算子,cache算子的底层调用的是persist算子…

网络安全风险评估

项目背景 随着信息化技术的快速发展,特别是面向社会、政府机构、企业等业务系统的投入使用,各组织机构对网络和信息系统安全防护都提出了新的要求。为满足安全需求,需对组织机构的网络和信息系统的安全进行一次系统全面的评估,以…

Selenium 自动化测试demo

场景描述: 模拟用户登录页面操作,包括输入用户名、密码、验证码。验证码为算数运算,如下: 使用到的工具和依赖: 1. Selenium:pip install selenium 2. 需要安装浏览器驱动:这里使用的是Edge 3…

架构-微服务架构

文章目录 前言一、系统架构演变1. 单体应用架构2. 垂直应用架构3. 分布式架构4. SOA 架构5. 微服务架构 二. 微服务架构介绍1. 微服务架构的常见问题2. 微服务架构的常见概念3. 微服务架构的常见解决方案4. 解决方案选型 三. Spring Cloud Alibaba介绍1. 主要功能2. 组件 前言 …

unity | 动画模块之卡片堆叠切换

一、预览动画 可以放很多图,可以自己往后加,可以调图片x轴和y轴间距,可以调图片飞出方向,可以调堆叠方向。 图1 图片堆叠动画预览 二、纯净代码 有粉丝问我这个效果,最近很忙,没有时间细写,先…

Burp入门(2)-代理功能介绍

声明:学习视频来自b站up主 泷羽sec,如涉及侵权马上删除文章 感谢泷羽sec 团队的教学 视频地址:burp功能介绍(1)_哔哩哔哩_bilibili 本文详细介绍burp的代理功能。 一、插件配置代理 1. 安装代理插件(Goog…

Jupyter Notebook的安装和配置提示功能

Python开发环境搭建conda管理环境-CSDN博客 安装anaconda和对接到编译器的教程可以看上面这一篇 Jupyter Notebook是一种交互式计算环境,它允许用户在单个文档中编写和执行代码、方程、可视化和文本。与其他编译器相比,Jupyter Notebook的突出点在于其交…

计算机网络 实验七 NAT配置实验

一、实验目的 通过本实验理解网络地址转换的原理和技术,掌握扩展NAT/NAPT设计、配置和测试。 二、实验原理 NAT配置实验的原理主要基于网络地址转换(NAT)技术,该技术用于将内部私有网络地址转换为外部公有网络地址,从…