首页 > 数据库 >Redis原子保种实现示例

Redis原子保种实现示例

来源：互联网 2026-07-10 08:43:01

核心逻辑：确保 Redis 中指定 key 存在，且其 TTL 不低于 60 秒；若 key 不存在，则通过 setnx 原子写入初始 seed 并设置 5 分钟过期时间。函数命名为 ensure_seed_with_ttl。以下提供带类型提示、英文日志与异常兜底的完整封装，同时指出两处值得优化的

核心逻辑：确保 Redis 中指定 key 存在，且其 TTL 不低于 60 秒；若 key 不存在，则通过 setnx 原子写入初始 seed 并设置 5 分钟过期时间。

Redis原子保种实现示例

长期稳定更新的攒劲资源： >>>点此立即查看<<<

函数命名为 ensure_seed_with_ttl。以下提供带类型提示、英文日志与异常兜底的完整封装，同时指出两处值得优化的细节，并给出改进后的代码。

一、原始逻辑直接封装

import datetime
import logging
from typing import Optional

log = logging.getLogger(__name__)

def ensure_seed_with_ttl(
    rediz,
    redis_key: str,
    item_type,
    msg: Optional[str] = None,
) -> None:
    """
    Ensure the Redis key exists and has a TTL of at least 60 s.
    If the key does not exist, atomically set it to the initial seed
    and give it a 5-minute expiration.
    """
    if not rediz.exists(redis_key):
        seed = ensure_seed_reference(item_type)
        result = rediz.setnx(redis_key, seed.ref_seq)
        log.info(
            "Seed ensured in Redis: result=%s, next_seq=%s, msg=%s",
            result, seed.ref_seq, msg
        )
    ttl = rediz.ttl(redis_key)
    if ttl < 60 or ttl == -1:          # -1 表示 key 存在但没有过期时间
        rediz.expire(redis_key, datetime.timedelta(minutes=5))

二、可优化的两点

这段代码可正常运行，但有两处可优化：

竞争窗口
两个客户端可能同时通过 exists 判断为 False，然后都执行 setnx。虽然 setnx 本身原子，但日志会打印两次“Seed ensured …”，容易造成误解。优化思路：将“判存在 + 设值 + 设过期”合并为一条原子操作（Lua 脚本或 Redis pipeline）。
减少一次 TTL 查询
若 setnx 成功，可直接设置过期时间为 5 分钟，省去后续 ttl 查询；仅在 setnx 失败（key 已存在）时检查 TTL。

三、优化后代码（Lua 脚本一次原子执行）

以下使用 Lua 脚本一次性解决上述两个问题：

import datetime
import logging
from typing import Optional

log = logging.getLogger(__name__)

# Lua 脚本：如果 key 不存在，则 set value 并设置 5 min 过期；
# 如果 key 已存在但 ttl<60，则延长到 5 min；
# 返回 1 表示写入了新 seed，返回 0 表示仅调整过期时间。
_ENSURE_SEED_LUA = """
local key, value, min_ttl, full_ttl = KEYS[1], ARGV[1], tonumber(ARGV[2]), tonumber(ARGV[3])
local exists = redis.call('exists', key)
if exists == 0 then
    redis.call('setnx', key, value)
    redis.call('expire', key, full_ttl)
    return 1
else
    local ttl = redis.call('ttl', key)
    if ttl < min_ttl then
        redis.call('expire', key, full_ttl)
    end
    return 0
end
"""

def ensure_seed_with_ttl(
    rediz,
    redis_key: str,
    item_type,
    msg: Optional[str] = None,
) -> None:
    """
    Atomically ensure the Redis key exists and has a TTL of at least 60 s.
    If the key does not exist, atomically set it to the initial seed
    and give it a 5-minute expiration.
    """
    seed = ensure_seed_reference(item_type)
    written = rediz.eval(
        _ENSURE_SEED_LUA, 
        1,                      # number of keys
        redis_key,
        seed.ref_seq,           # ARGV[1] 
        60,                     # ARGV[2]  min_ttl
        int(datetime.timedelta(minutes=5).total_seconds())  # ARGV[3]  full_ttl
    )
    if written == 1:
        log.info(
            "Seed ensured in Redis: next_seq=%s, msg=%s",
            seed.ref_seq, msg
        )

优化效果：