如何编写定时刷新缓存的脚本

wen 实用脚本 2026-08-01 2

本文目录导读：

如何编写定时刷新缓存的脚本

Python方案（最灵活）
Bash Shell脚本方案
Node.js方案
使用Cron定时任务
Docker容器化方案
使用建议

Python方案（最灵活）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import redis
import time
import logging
from datetime import datetime
import json
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('cache_refresh.log'),
        logging.StreamHandler()
    ]
)
class CacheRefresher:
    def __init__(self):
        # Redis连接配置
        self.redis_client = redis.Redis(
            host='localhost',
            port=6379,
            db=0,
            decode_responses=True
        )
        # 需要刷新的缓存键配置
        self.cache_configs = [
            {
                'key': 'user:list',
                'source': 'database',  # 数据来源
                'refresh_interval': 300,  # 刷新间隔（秒）
                'last_refresh': 0
            },
            {
                'key': 'product:hot',
                'source': 'api',
                'refresh_interval': 600,
                'last_refresh': 0
            }
        ]
    def fetch_data(self, source):
        """模拟从数据源获取数据"""
        if source == 'database':
            # 这里替换为实际的数据库查询
            return {
                'timestamp': datetime.now().isoformat(),
                'data': [{'id': 1, 'name': '测试数据'}]
            }
        elif source == 'api':
            # 这里替换为实际的API调用
            return {
                'timestamp': datetime.now().isoformat(),
                'data': {'status': 'ok', 'items': []}
            }
        return None
    def refresh_cache(self):
        """刷新缓存"""
        current_time = time.time()
        for config in self.cache_configs:
            # 检查是否需要刷新
            if current_time - config['last_refresh'] >= config['refresh_interval']:
                try:
                    logging.info(f"开始刷新缓存: {config['key']}")
                    # 获取新数据
                    new_data = self.fetch_data(config['source'])
                    # 更新缓存
                    self.redis_client.setex(
                        config['key'],
                        config['refresh_interval'] * 2,  # 设置较长的过期时间
                        json.dumps(new_data)
                    )
                    # 更新刷新时间
                    config['last_refresh'] = current_time
                    logging.info(f"缓存刷新成功: {config['key']}")
                except Exception as e:
                    logging.error(f"刷新缓存失败 {config['key']}: {str(e)}")
    def run(self):
        """运行主循环"""
        logging.info("缓存刷新服务启动")
        try:
            while True:
                self.refresh_cache()
                time.sleep(1)  # 每秒检查一次
        except KeyboardInterrupt:
            logging.info("缓存刷新服务停止")
if __name__ == "__main__":
    refresher = CacheRefresher()
    refresher.run()

Bash Shell脚本方案

#!/bin/bash
# 定时刷新缓存的Shell脚本
# 配置
REDIS_HOST="localhost"
REDIS_PORT="6379"
LOG_FILE="/var/log/cache_refresh.log"
TEMP_DIR="/tmp/cache_temp"
# 创建临时目录
mkdir -p "$TEMP_DIR"
# 日志函数
log() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" >> "$LOG_FILE"
}
# 刷新Redis缓存
refresh_redis_cache() {
    local key="$1"
    local value="$2"
    local expire="$3"
    # 使用redis-cli设置缓存
    redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" SETEX "$key" "$expire" "$value"
    if [ $? -eq 0 ]; then
        log "成功刷新缓存: $key"
    else
        log "刷新缓存失败: $key"
    fi
}
# 从数据库获取数据（示例）
fetch_from_database() {
    # 这里替换为实际的数据库查询
    echo "{\"timestamp\":\"$(date -Iseconds)\",\"data\":\"示例数据\"}"
}
# 从API获取数据（示例）
fetch_from_api() {
    # 这里替换为实际的API调用
    curl -s http://api.example.com/data 2>/dev/null || echo "{\"error\":\"API不可用\"}"
}
# 主刷新逻辑
main_refresh() {
    log "开始刷新缓存任务"
    # 刷新使用频率高的缓存
    user_data=$(fetch_from_database)
    refresh_redis_cache "user:list" "$user_data" 600
    # 刷新热点数据
    hot_data=$(fetch_from_api)
    refresh_redis_cache "product:hot" "$hot_data" 900
    log "刷新缓存任务完成"
    echo "缓存刷新完成"
}
# 根据参数执行任务
case "${1:-run}" in
    run)
        main_refresh
        ;;
    loop)
        # 循环模式
        INTERVAL="${2:-300}"  # 默认300秒刷新一次
        log "启动循环刷新模式，间隔：${INTERVAL}秒"
        while true; do
            main_refresh
            sleep "$INTERVAL"
        done
        ;;
    *)
        echo "用法: $0 {run|loop [间隔秒]}"
        exit 1
        ;;
esac

Node.js方案

// cache-refresher.js
const Redis = require('ioredis');
const schedule = require('node-schedule');
const logger = require('winston');
// 配置日志
logger.configure({
    transports: [
        new logger.transports.File({ filename: 'cache-refresh.log' }),
        new logger.transports.Console()
    ]
});
class CacheRefresher {
    constructor() {
        this.redis = new Redis({
            host: 'localhost',
            port: 6379
        });
        // 缓存配置
        this.cacheConfigs = [
            {
                name: 'userCache',
                key: 'user:list',
                cron: '*/5 * * * *',  // 每5分钟
                fetchData: this.fetchUsers.bind(this)
            },
            {
                name: 'productCache',
                key: 'product:hot',
                cron: '*/10 * * * *',  // 每10分钟
                fetchData: this.fetchHotProducts.bind(this)
            }
        ];
    }
    // 模拟获取用户数据
    async fetchUsers() {
        // 这里替换为实际的数据库查询
        return {
            timestamp: new Date().toISOString(),
            data: [{ id: 1, name: '用户1' }, { id: 2, name: '用户2' }]
        };
    }
    // 模拟获取热门商品
    async fetchHotProducts() {
        // 这里替换为实际的API调用
        return {
            timestamp: new Date().toISOString(),
            data: [{ id: 'P001', sales: 100 }]
        };
    }
    // 刷新单个缓存
    async refreshCache(config) {
        try {
            logger.info(`开始刷新缓存: ${config.name}`);
            // 获取新数据
            const data = await config.fetchData();
            // 更新缓存（设置2小时的过期时间）
            await this.redis.setex(
                config.key,
                7200,
                JSON.stringify(data)
            );
            logger.info(`缓存刷新成功: ${config.name}`);
        } catch (error) {
            logger.error(`缓存刷新失败 ${config.name}: ${error.message}`);
        }
    }
    // 启动定时任务
    start() {
        logger.info('缓存刷新服务启动');
        this.cacheConfigs.forEach(config => {
            // 使用node-schedule设置定时任务
            schedule.scheduleJob(config.cron, () => {
                this.refreshCache(config);
            });
        });
    }
    // 手动触发刷新
    async manualRefresh(key) {
        const config = this.cacheConfigs.find(c => c.key === key);
        if (config) {
            await this.refreshCache(config);
            return `已刷新缓存: ${key}`;
        }
        return `未找到缓存配置: ${key}`;
    }
    // 停止服务
    async stop() {
        logger.info('缓存刷新服务停止');
        await this.redis.quit();
    }
}
// 启动服务
const refresher = new CacheRefresher();
refresher.start();
// 保持进程运行
process.on('SIGINT', async () => {
    await refresher.stop();
    process.exit(0);
});

使用Cron定时任务

创建cache_refresh.sh脚本后,添加到crontab：

# 编辑crontab
crontab -e
# 添加以下配置：
# 每5分钟刷新一次
*/5 * * * * /usr/local/bin/cache_refresh.sh run >> /var/log/cache_cron.log 2>&1
# 每小时的第10分钟刷新
10 * * * * /usr/local/bin/cache_refresh.sh special_refresh >> /var/log/cache_cron.log 2>&1
# 每天凌晨2点刷新
0 2 * * * /usr/local/bin/cache_refresh.sh daily_refresh >> /var/log/cache_cron.log 2>&1

Docker容器化方案

# Dockerfile
FROM python:3.9-slim
WORKDIR /app
# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制脚本
COPY cache_refresher.py .
# 运行脚本
CMD ["python", "cache_refresher.py"]

使用建议

选择合适的刷新策略：
- 定时刷新：适合数据更新频率低的情况
- 手动触发：适合数据变化可预测的场景
- 事件驱动：适合数据实时性要求高的场景

缓存一致性处理：

# 使用版本号或时间戳
def refresh_with_version():
    version = get_data_version()
    key = f"cache:{entity}:{version}"
    set_cache(key, data)

错误处理和重试机制：

import time
def refresh_with_retry(func, max_retries=3):
    for i in range(max_retries):
        try:
            return func()
        except Exception as e:
            if i == max_retries - 1:
                raise
            time.sleep(2 ** i)  # 指数退避

监控和告警：

def check_refresh_status():
    # 检查最近一次刷新时间
    last_refresh = get_last_refresh_time()
    if time.time() - last_refresh > threshold:
        send_alert("缓存刷新异常")

选择合适的方案部署时，请根据实际业务需求调整刷新频率、数据源连接和错误处理逻辑。