Redis 原子保种的实现示例

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))

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
        )