Redis进阶实战：香港服务器实现分布式锁 + 限流 + 排行榜 + 秒杀库存扣减完整方案

Redis 不只是缓存工具。凭借原子操作、Lua 脚本和丰富的数据结构，Redis 可以优雅解决分布式系统中的多种高频问题。本文通过四个真实业务场景，展示 Redis 在香港服务器生产环境中的进阶用法。

一、分布式锁（防止重复下单/重复支付）

在多进程/多服务器环境中，若不加锁，同一用户可能同时提交两个请求都通过了库存检查，导致超卖。分布式锁确保同一时刻只有一个进程能操作同一资源。

<code">import redis
import uuid
import time
from contextlib import contextmanager

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

@contextmanager
def distributed_lock(lock_key: str, expire_seconds: int = 10, retry_times: int = 3):
    """
    分布式锁（SET NX + 随机值防误解锁）
    用法：
        with distributed_lock(f"order:{order_id}"):
            # 处理订单
    """
    lock_value = str(uuid.uuid4())   # 唯一值，防止其他进程误解锁
    acquired = False

    for attempt in range(retry_times):
        # SET key value NX EX（原子操作：仅在 key 不存在时设置）
        acquired = r.set(
            lock_key,
            lock_value,
            nx=True,         # Not Exists：只在 key 不存在时设置
            ex=expire_seconds # 自动过期防止死锁
        )
        if acquired:
            break
        time.sleep(0.1 * (attempt + 1))   # 指数退避重试

    if not acquired:
        raise Exception(f"无法获取锁: {lock_key}，请稍后重试")

    try:
        yield
    finally:
        # Lua 脚本原子解锁（只解自己的锁，防止解锁他人的锁）
        unlock_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        r.eval(unlock_script, 1, lock_key, lock_value)

# 使用示例：防止同一订单被重复支付
def process_payment(order_id: str, amount: float):
    lock_key = f"payment_lock:{order_id}"
    with distributed_lock(lock_key, expire_seconds=30):
        # 锁内操作：检查订单状态 → 扣款 → 更新状态
        order = get_order(order_id)
        if order['status'] == 'paid':
            raise Exception("订单已支付")
        charge_card(order_id, amount)
        update_order_status(order_id, 'paid')

二、滑动窗口限流（API 调用频率控制）

<code">def sliding_window_rate_limit(
    user_id: str,
    limit: int = 100,
    window_seconds: int = 60
) -> tuple[bool, int]:
    """
    滑动窗口限流（精确，无固定窗口的边界突刺问题）
    返回：(是否允许, 剩余可用次数)
    """
    key = f"rate_limit:{user_id}"
    now = time.time()
    window_start = now - window_seconds

    pipe = r.pipeline()
    # 移除窗口外的旧记录
    pipe.zremrangebyscore(key, 0, window_start)
    # 添加当前请求（score = 时间戳）
    pipe.zadd(key, {str(uuid.uuid4()): now})
    # 统计窗口内请求数
    pipe.zcard(key)
    # 设置过期时间（自动清理）
    pipe.expire(key, window_seconds + 1)
    results = pipe.execute()

    current_count = results[2]
    remaining = max(0, limit - current_count)

    if current_count > limit:
        # 超限，移除刚才添加的记录
        r.zremrangebyscore(key, now, now)
        return False, 0

    return True, remaining

# FastAPI 中间件示例
from fastapi import Request, HTTPException

async def rate_limit_middleware(request: Request, call_next):
    user_id = request.headers.get("X-User-ID", request.client.host)
    allowed, remaining = sliding_window_rate_limit(user_id, limit=100, window_seconds=60)

    if not allowed:
        raise HTTPException(
            status_code=429,
            detail="请求过于频繁，请 60 秒后重试",
            headers={"X-RateLimit-Remaining": "0", "Retry-After": "60"}
        )

    response = await call_next(request)
    response.headers["X-RateLimit-Remaining"] = str(remaining)
    return response

三、实时排行榜（电商销量/积分榜）

<code">class RealTimeRanking:
    """基于 Redis Sorted Set 的实时排行榜"""

    def __init__(self, board_name: str, expire_days: int = 30):
        self.key = f"ranking:{board_name}"
        self.expire_seconds = expire_days * 86400

    def add_score(self, member: str, score: float):
        """增加分数（销量+1、积分+N 等）"""
        r.zadd(self.key, {member: score}, incr=True)
        r.expire(self.key, self.expire_seconds)

    def set_score(self, member: str, score: float):
        """设置绝对分数"""
        r.zadd(self.key, {member: score})

    def get_rank(self, member: str) -> dict:
        """获取某成员的排名和分数"""
        rank = r.zrevrank(self.key, member)      # 从高到低排名（0-indexed）
        score = r.zscore(self.key, member)
        if rank is None:
            return {"rank": None, "score": 0}
        return {"rank": rank + 1, "score": score}

    def get_top(self, top_n: int = 10) -> list[dict]:
        """获取 Top N 排行榜"""
        results = r.zrevrange(self.key, 0, top_n - 1, withscores=True)
        return [
            {"rank": i + 1, "member": member, "score": score}
            for i, (member, score) in enumerate(results)
        ]

    def get_around(self, member: str, radius: int = 5) -> list[dict]:
        """获取某成员周围的排名（竞品分析）"""
        rank = r.zrevrank(self.key, member)
        if rank is None:
            return []
        start = max(0, rank - radius)
        end = rank + radius
        results = r.zrevrange(self.key, start, end, withscores=True)
        return [
            {"rank": start + i + 1, "member": m, "score": s, "is_self": m == member}
            for i, (m, s) in enumerate(results)
        ]

# 使用示例
daily_sales = RealTimeRanking("daily_sales:2026-07-16")
daily_sales.add_score("product:123", 1)     # 商品卖出1件
daily_sales.add_score("product:456", 3)
top10 = daily_sales.get_top(10)
my_rank = daily_sales.get_rank("product:123")

四、秒杀库存原子扣减（防超卖）

<code">class FlashSaleInventory:
    """秒杀库存管理（Redis 原子操作，绝对不超卖）"""

    def __init__(self, product_id: str):
        self.stock_key = f"flash_sale:stock:{product_id}"
        self.sold_key  = f"flash_sale:sold:{product_id}"

    def init_stock(self, quantity: int):
        """活动开始前初始化库存"""
        pipe = r.pipeline()
        pipe.set(self.stock_key, quantity)
        pipe.delete(self.sold_key)
        pipe.execute()

    # Lua 脚本：原子检查并扣减库存（在 Redis 内部执行，不存在并发问题）
    DEDUCT_SCRIPT = """
    local stock = tonumber(redis.call('get', KEYS[1]))
    local qty   = tonumber(ARGV[1])
    local order_id = ARGV[2]

    if stock == nil then
        return {-2, '活动未初始化'}
    end

    if stock < qty then return {-1, stock} end redis.call('decrby', KEYS[1], qty) redis.call('hset', KEYS[2], order_id, qty) return {1, stock - qty} """ def deduct(self, quantity: int, order_id: str) -> dict:
        """
        原子扣减库存
        返回：{"success": True/False, "remaining": 剩余库存, "message": ""}
        """
        result = r.eval(
            self.DEDUCT_SCRIPT,
            2,                       # 2 个 KEY
            self.stock_key,
            self.sold_key,
            quantity,
            order_id
        )
        code, value = result
        if code == 1:
            return {"success": True, "remaining": int(value)}
        elif code == -1:
            return {"success": False, "remaining": int(value), "message": "库存不足"}
        else:
            return {"success": False, "remaining": 0, "message": str(value)}

    def get_stock(self) -> int:
        """查询当前库存"""
        val = r.get(self.stock_key)
        return int(val) if val else 0

# 秒杀接口示例
flash = FlashSaleInventory("product:iphone16")
flash.init_stock(100)    # 活动开始，初始化100件库存

# 用户抢购（并发安全）
result = flash.deduct(1, order_id="ord_abc123")
if result["success"]:
    # 创建订单、等待支付
    create_pending_order(order_id, "product:iphone16", 1)
else:
    raise Exception(f"抢购失败: {result['message']}")

五、Redis 内存优化配置

<code"># /etc/redis/redis.conf 生产环境关键参数

# 最大内存（根据服务器内存调整，留 20% 给系统）
maxmemory 6gb
maxmemory-policy allkeys-lru   # 内存满时淘汰最近最少使用

# 持久化策略（AOF + RDB 双重保障）
appendonly yes
appendfsync everysec            # 每秒刷盘（性能与安全的平衡）
save 900 1                      # 900秒内有1次写操作则 RDB 持久化
save 300 10

# 慢查询日志（排查性能问题）
slowlog-log-slower-than 10000  # 超过 10ms 的命令记录
slowlog-max-len 128

# 禁用危险命令（生产安全）
rename-command FLUSHDB  ""
rename-command FLUSHALL ""
rename-command CONFIG   "CONFIG_SAFE_CMD"

六、总结

Redis 的原子操作和 Lua 脚本让分布式系统中的「检查-操作」问题得到优雅解决。分布式锁防止重复支付、滑动窗口限流保护 API 稳定性、ZSet 排行榜实时统计销量、Lua 脚本原子扣减杜绝超卖——这四个模式覆盖了电商系统最核心的 Redis 使用场景，可直接在香港 VPS 生产环境中应用。