Redis原子保种的实现示例

功能一句话概括：
“确保 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 < 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 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<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 )

优化效果

到此这篇关于Redis 原子保种的实现示例的文章就介绍到这了,