Redis 缓存数据频繁更新导致的性能开销及优化策略

import redis
from datetime import datetime

r = redis.Redis()

def delayed_update(key, value, delay=5):
    """缓冲5秒内的更新请求"""
    current_time = datetime.now().timestamp()
    # 使用有序集合存储时间戳作为分数
    r.zadd(f"buffer:{key}", {value: current_time})
    # 检查是否达到批量处理阈值
    if r.zcard(f"buffer:{key}") > 50:
        process_buffer(key)

def process_buffer(key):
    """处理缓冲数据并更新缓存"""
    # 获取最近5秒内的所有值
    values = r.zrangebyscore(f"buffer:{key}", datetime.now().timestamp()-5, '+inf')
    # 取最后一个有效值（类似防抖逻辑）
    if values:
        r.set(key, values[-1])
    # 清理已处理数据
    r.delete(f"buffer:{key}")

const Redis = require('ioredis');
const redis = new Redis();

async function batchUpdateProducts(products) {
    const pipeline = redis.pipeline();
    products.forEach(({id, stock}) => {
        pipeline.hset(`product:${id}`, 'stock', stock);
    });
    // 单次网络往返完成所有操作
    await pipeline.exec(); 
    
    // 错误处理示例
    pipeline.exec().then(results => {
        results.forEach(([err, res], index) => {
            if(err) console.error(`更新失败: ${products[index].id}`);
        });
    });
}

// Spring Data Redis示例
@CachePut(value = "userProfile", key = "#userId")
public UserProfile updateProfile(String userId, UserProfile profile) {
    // 动态设置过期时间（1小时基础 + 随机30分钟）
    redisTemplate.expire("userProfile::"+userId, 
        60*60 + ThreadLocalRandom.current().nextInt(0, 1800), 
        TimeUnit.SECONDS);
    return profileRepository.save(profile);
}

# Celery + Redis实现异步队列
@app.task
def async_cache_update(key, data):
    try:
        with redis.pipeline() as pipe:
            pipe.set(key, json.dumps(data))
            pipe.expire(key, 3600)
            pipe.execute()
    except Exception as e:
        logger.error(f"缓存更新失败: {key} - {str(e)}")
        async_cache_update.retry(countdown=30)

# 业务代码调用
async_cache_update.delay("hot_product_list", new_products)