如何用Python在Redis中批量实现获取或设置缓存（不存在则更新）

需要实现Redis多键的获取与设置缓存功能，当键不存在时需更新缓存。目前已实现单键处理函数，但该函数每次仅能处理一个键值对，希望优化为通过一次Redis连接批量处理多键，同时保留原有逻辑：

• 若键存在则返回缓存值；
• 若不存在，若default为函数则执行函数获取结果并存入缓存，否则直接设置默认值。

现有单键处理代码：

import os
from redis.client import Redis

REDIS_HOST = os.environ["REDIS_HOST"]
REDIS_PORT = int(os.environ.get("REDIS_PORT", 6379))

redis_client = Redis(host=REDIS_HOST, port=REDIS_PORT, db=0)

DEFAULT_CACHE_TIMEOUT = int(os.environ.get("DEFAULT_CACHE_TIMEOUT", 60 * 60 * 24))


def get_or_set_cache(key, default, timeout=DEFAULT_CACHE_TIMEOUT, function_args=None, function_kwargs=None):
    # type: (str, any, int, *any, **any) -> any
    """Get value from cache or set default value to cache and return it"""
    cached_result = redis_client.get(key)
    if cached_result:
        return cached_result

    if type(default).__name__ == 'function':
        result = default(*function_args or (), **function_kwargs or {})
        redis_client.set(key, result, timeout)
        return result
    else:
        redis_client.set(key, default, timeout)
        return default

已查阅资料了解Redis批量获取/设置方法，但缺失键不存在时的缓存设置逻辑，求实现方案。

解决方案 #

核心思路 #

1. 批量获取缓存：用Redis的mget方法一次性拉取所有键的缓存值，减少连接开销；
2. 筛选缺失键：从批量结果中定位出不存在的键及其位置；
3. 批量填充缺失值：针对缺失键，根据default类型（静态值/函数）生成对应结果，再通过Pipeline批量写入Redis（优化多次set的交互成本）；
4. 整合结果：将已有缓存值与新生成值按原键顺序合并返回。

实现代码 #

import os
from redis.client import Redis

REDIS_HOST = os.environ["REDIS_HOST"]
REDIS_PORT = int(os.environ.get("REDIS_PORT", 6379))

redis_client = Redis(host=REDIS_HOST, port=REDIS_PORT, db=0)

DEFAULT_CACHE_TIMEOUT = int(os.environ.get("DEFAULT_CACHE_TIMEOUT", 60 * 60 * 24))


def get_or_set_cache_single(key, default, timeout=DEFAULT_CACHE_TIMEOUT, function_args=None, function_kwargs=None):
    """保留原单键逻辑，支持单独调用或被批量函数复用"""
    cached_result = redis_client.get(key)
    if cached_result:
        return cached_result

    if callable(default):
        result = default(*function_args or (), **function_kwargs or {})
        redis_client.set(key, result, timeout)
        return result
    else:
        redis_client.set(key, default, timeout)
        return default


def get_or_set_cache_batch(keys, default, timeout=DEFAULT_CACHE_TIMEOUT, function_args_list=None, function_kwargs_list=None):
    # type: (list[str], any, int, list[any], list[dict]) -> list[any]
    """批量获取或设置缓存，返回结果顺序与输入keys完全一致"""
    # 1. 批量获取已有缓存
    cached_results = redis_client.mget(keys)
    
    # 2. 标记缺失的键及其索引
    missing_keys = []
    missing_indices = []
    for idx, (key, val) in enumerate(zip(keys, cached_results)):
        if val is None:
            missing_keys.append(key)
            missing_indices.append(idx)
    
    # 3. 生成缺失键的值并批量写入缓存
    new_values = []
    if missing_keys:
        # 为每个缺失键准备函数参数，默认空参数，支持按键传参
        func_args = function_args_list or [()] * len(missing_keys)
        func_kwargs = function_kwargs_list or [{}] * len(missing_keys)
        
        # 生成对应值
        for idx, key in enumerate(missing_keys):
            if callable(default):
                val = default(*func_args[idx], **func_kwargs[idx])
            else:
                val = default
            new_values.append(val)
        
        # 用Pipeline批量执行set，减少Redis交互次数
        with redis_client.pipeline() as pipe:
            for key, val in zip(missing_keys, new_values):
                pipe.set(key, val, timeout)
            pipe.execute()
    
    # 4. 合并结果，保持原键顺序
    final_results = list(cached_results)
    for idx, val in zip(missing_indices, new_values):
        final_results[idx] = val
    
    return final_results

关键说明 #

• 保留原单键函数get_or_set_cache_single，兼容原有单键使用场景；
• 批量函数支持：
  • 静态default值：所有缺失键统一使用该值填充；
  • 函数型default：为每个缺失键执行函数生成值，支持通过function_args_list和function_kwargs_list为不同键传入差异化参数；
  • 用Pipeline优化批量写入操作，避免多次Redis连接开销；
  • 返回结果顺序与输入keys严格对应，无需额外匹配键值对。

内容的提问来源于stack exchange，提问作者RobertPro