功能一句话概括：
“确保 Redis 里存在该 key，且其 TTL ≥ 60 s，否则用 setnx 写入初始 seed 并把过期时间设为 5 min。”
据此可取名：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
二、可优化的两点
-
竞争窗口
两个客户端可能同时通过exists判断为 False，然后都执行setnx，虽然setnx本身是原子性的，但日志会打印两次“Seed ensured …”，容易误导。
优化：把“判存在 + 设值 + 设过期”做成一条原子事务（Lua 脚本或 Redis pipeline）。 -
减少一次 TTL 查询
如果setnx成功，我们可以直接把过期时间设为 5 min，省去一次ttl查询；
只有setnx失败（key 已存在）时才需要ttl检查。
三、优化后代码（Lua 脚本一次原子执行）
import datetime
import logging
from typing import Optional
log = logging.getLogger(__name__)
# Lua 脚本：如果 key 不存在，则 set value 并设置 5 min 过期；
# 如果 key 已存在但 ttl 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
)
优化效果
- 竞争窗口消失，全程原子；
- 减少一次
ttl网络往返（在 key 不存在场景）； - 日志只在真正写入 seed 时打印一次，避免重复误导。
到此这篇关于Redis 原子保种的实现示例的文章就介绍到这了,更多相关Redis 原子保种内容请搜索IT俱乐部以前的文章或继续浏览下面的相关文章希望大家以后多多支持IT俱乐部！
