快捷導(dǎo)航

Redis 原子保種的實(shí)現(xiàn)示例

更新時(shí)間：2026年03月31日 10:33:22 作者：Leon_Jinhai_Sun

本文主要介紹了Redis 原子保種的實(shí)現(xiàn)示例,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

一、原始邏輯直接封裝

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 存在但沒(méi)有過(guò)期時(shí)間
        rediz.expire(redis_key, datetime.timedelta(minutes=5))

二、可優(yōu)化的兩點(diǎn)

競(jìng)爭(zhēng)窗口
兩個(gè)客戶端可能同時(shí)通過(guò) exists 判斷為 False，然后都執(zhí)行 setnx，雖然 setnx 本身是原子性的，但日志會(huì)打印兩次“Seed ensured …”，容易誤導(dǎo)。
優(yōu)化：把“判存在 + 設(shè)值 + 設(shè)過(guò)期”做成一條原子事務(wù)（Lua 腳本或 Redis pipeline）。
減少一次 TTL 查詢
如果 setnx 成功，我們可以直接把過(guò)期時(shí)間設(shè)為 5 min，省去一次 ttl 查詢；
只有 setnx 失?。╧ey 已存在）時(shí)才需要 ttl 檢查。

三、優(yōu)化后代碼（Lua 腳本一次原子執(zhí)行）

import datetime
import logging
from typing import Optional
log = logging.getLogger(__name__)
# Lua 腳本：如果 key 不存在，則 set value 并設(shè)置 5 min 過(guò)期；
# 如果 key 已存在但 ttl<60，則延長(zhǎng)到 5 min；
# 返回 1 表示寫(xiě)入了新 seed，返回 0 表示僅調(diào)整過(guò)期時(shí)間。
_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
        )

優(yōu)化效果