分布式微服务系统架构第169集：1万~10万QPS的查当前订单列表

发布于：2025-09-06 ⋅ 阅读:(14) ⋅ 点赞:(0)

加群联系作者vx：xiaoda0423

仓库地址：https://webvueblog.github.io/JavaPlusDoc/

https://1024bat.cn/

https://github.com/webVueBlog/fastapi_plus

https://webvueblog.github.io/JavaPlusDoc/

点击勘误issues，哪吒感谢大家的阅读

如果目标是“在 Redis 上把读流量兜住、即使 Redis 未命中也不把洪峰直接打到 Mongo”，可以把“防穿透/合并/限流/热键打散/负缓存/失效广播”全放到 Redis 层完结。下面给出可落地的 Redis 侧方案（键模型 + Lua + Pub/Sub/Stream + 用法示例）。

1) 键模型（用 hash-tag 保证相关键同槽）

这样在 Redis Cluster 里多键原子操作也能跑（Lua 限同槽）。

  • 当前订单映射（用户→进行中订单ID或空）
    cur:{op}:{uid} -> orderId | "-"（TTL 120–300s，±10% 抖动）

  • 订单详情
    od:{orderId} -> JSON（TTL 300–600s，±10%）

  • 订单列表索引（分页）
    olist:{op}:{uid} -> ZSET(ts -> orderId)（可选）

  • 热点复制（打散热 key）
    cur:{op}:{uid}#1..N（读随机 1..N，写 N 份）

  • 互斥闸（只允许一个回源）
    gate:{op}:{uid}（短 TTL 2–5s）

  • 失效广播（清 L1 或提醒客户端）
    PUBLISH inv:{op}:{uid} <payload>

  • 回源任务（可异步处理）
    XADD miss:cur * key cur:{op}:{uid}

{} 里的内容做 hash-tag，例如 cur:{op123:uid456}，确保同槽。

2) 纯 Redis 原子逻辑（Lua）

2.1 读取当前订单（带“单飞闸 + 负缓存”）

效果：

  • 命中直接返回

  • 未命中时抢闸（只有 1 个客户端拿到），其他立即收到 _MISS_LOCKED_，不去 Mongo

  • 支持热点复制读取

-- KEYS[1] = cur key (或某个副本key)
-- KEYS[2] = gate key
-- ARGV[1] = gate ttl ms
local v = redis.call('GET', KEYS[1])
if v then
  return v  -- 命中：orderId 或 "-"
end
-- 未命中：尝试抢闸
local ok = redis.call('SET', KEYS[2], '1', 'NX', 'PX', ARGV[1])
if ok then
  return '_MISS_NEED_FILL_'  -- 只有拿到闸的人去“回源/异步填充”
else
  return '_MISS_LOCKED_'     -- 别的请求立即返回，不打库
end

Java 调用示例（选随机副本读取）：

String baseKey = "cur:{%s:%s}".formatted(op, uid);
String curKey = baseKey + "#" + (ThreadLocalRandom.current().nextInt(3) + 1);
String gateKey = "gate:{%s:%s}".formatted(op, uid);
String res = stringRedisTemplate.execute(scriptGetWithGate, List.of(curKey, gateKey), "3000");

switch (res) {
  case "-":    return null;            // 负缓存：无当前订单
  case "_MISS_NEED_FILL_":
    // 你可以：1) 轻量返回旧值/提示稍后  2) XADD miss:cur 交给异步去填充  3) 少量同步回源
    // 这里推荐：写入 Stream，再立即返回（保护 Mongo）
    stringRedisTemplate.opsForStream().add("miss:cur", Map.of("key", baseKey));
    return null;
  case "_MISS_LOCKED_":
    // 不打库，快速返回（或短暂自旋再查一次）
    return null;
  default:
    return "NULL".equals(res) || "-".equals(res) ? null : fetchDetail(res);
}

要极致一致也可以让“拿到闸”的那一个同步回源；但高峰期建议走 Stream 异步（见 §3）。

2.2 一次性写入“当前订单 + 详情 + N 份热 key 副本”（原子）

效果：订单创建/状态变化时，只打一把脚本，完成 MSET + EXPIRE + 广播。

-- KEYS: cur#1..N, detailKey, invChannel, baseCurKey
-- ARGV: orderId, ttlCurSec, ttlDetSec
local orderId = ARGV[1]
local ttlCur = tonumber(ARGV[2])
local ttlDet = tonumber(ARGV[3])

-- 写 N 份 cur 副本
for i=1,#KEYS-3 do
  redis.call('SET', KEYS[i], orderId, 'EX', ttlCur + math.random(0,ttlCur/10))
end
-- 写详情（这里假设上层已先把 JSON 放入临时键，或直接传 JSON 用 EVALSHA 限长注意）
redis.call('SET', KEYS[#KEYS-2], redis.call('GET', KEYS[#KEYS-2]) or '', 'EX', ttlDet + math.random(0,ttlDet/10))
-- 广播无论如何发一下，提醒 L1 失效/刷新
redis.call('PUBLISH', KEYS[#KEYS-1], KEYS[#KEYS])
return 'OK'

也可以把负缓存落盘：把 orderId 改成 "-"，TTL 短一点（30–60s）。

2.3 令牌桶（全在 Redis，限制落库速率）

效果：只有拿到令牌的人才允许“回源 Mongo”，否则返回降级。

-- KEYS[1] = bucket key
-- ARGV[1] = capacity
-- ARGV[2] = refill tokens per second
-- ARGV[3] = now (ms)
-- ARGV[4] = tokens to take
local cap = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local need = tonumber(ARGV[4])

local lastTokens = tonumber(redis.call('HGET', KEYS[1], 'tokens') or cap)
local lastRefill = tonumber(redis.call('HGET', KEYS[1], 'ts') or now)
local delta = math.max(0, now - lastRefill) / 1000.0 * rate
local current = math.min(cap, lastTokens + delta)
if current < need then
  redis.call('HSET', KEYS[1], 'tokens', current, 'ts', now)
  redis.call('EXPIRE', KEYS[1], 60)
  return 0
else
  redis.call('HSET', KEYS[1], 'tokens', current - need, 'ts', now)
  redis.call('EXPIRE', KEYS[1], 60)
  return 1
end

Java 侧：

boolean canHitDB = Boolean.TRUE.equals(
  stringRedisTemplate.execute(tokenBucketScript,
    List.of("rl:dbFallback"),
    "2000", "500", String.valueOf(System.currentTimeMillis()), "1"));
if (!canHitDB) return null; // 快速降级，不落库

3) 用 Redis 异步回源（不“打 Mongo 洪峰”）

3.1 生产“回源任务”

未命中且拿到闸的请求：

XADD miss:cur * key "cur:{op:uid}"

3.2 消费“回源任务”（多 worker 竞消费）

XGROUP CREATE miss:cur g1 $ MKSTREAM   # 初始化一次
XREADGROUP GROUP g1 c1 COUNT 100 BLOCK 2000 STREAMS miss:cur >
# worker 处理：查 Mongo → set cur/od → PUBLISH inv:... → XACK

这样回源的并发由 XREADGROUP 的worker 数量控制，不随前台流量放大；查库速率再叠加上面的令牌桶更稳。

4) 热点 Key 打散（全在 Redis）

  • 读：随机读 cur#1..N（或本机一致性哈希选一份）

  • 写：Lua 一次写 N 份（见 §2.2）

  • 删/失效：同样脚本批量 DEL/EXPIRE + PUBLISH

N 取 3~5 即可，能大幅摊薄单 key QPS。

5) 列表页也“只在 Redis 处理”

  • 用户“我的订单列表”：

    • ZREVRANGE olist:{op}:{uid} offset count 拿一页 orderId

    • 一次 MGET/Pipeline 拉取 od:{orderId...}

    • 缺失的用低频异步补全（Stream），前台先返回已有的

列表长期数据可只放“最近 N 条”在 Redis，老数据分页再走后端“离线索引 + 异步补全”。

6) 失效广播（Redis Pub/Sub）

  • 写路径变更后：PUBLISH inv:{op}:{uid} <orderId|->

  • 应用订阅 inv:*，拿到消息就清 L1或触发局部刷新

  • 这样避免 L2 改了、L1 还留旧值导致命中错误

7) 策略小抄（全靠 Redis 就能做到的）

  • 负缓存：用 "-"/"NULL" 短 TTL，防止穿透

  • 随机 TTL：±10–20% 抖动，避同刻雪崩

  • 单飞互斥：短期 SET NX PX 或 RedLock（倾向单实例短锁即可）

  • 令牌桶限流：Lua（§2.3）限制允许回源的频率

  • 热点复制：#1..N

  • 两段式检查（双检） ：拿闸后再 GET一次，避免重复回源

  • Pipeline/MGET/MSET：批量 IO 降 RTT

  • Key 空间规划：统一前缀 + hash-tag，便于 Cluster 同槽操作

  • 淘汰策略：allkeys-lru 或 volatile-lru；给所有缓存都设 TTL

8) 最小落地清单

  1. 采用上面的 键模型；

  2. 接入 Lua 读闸脚本 + 令牌桶脚本；

  3. 读路径：L1→L2→闸；未命中XADD并快速返回；

  4. 后台 worker：XREADGROUP 拉取 miss → 回源 → 批量写回 Lua → PUBLISH；

  5. 写路径：订单状态变化走批量写回 Lua（同时发广播）；

  6. Redis Cluster：给相关键加 hash-tag；热键做 #1..N；

  7. 监控：命中率、闸命中率、回源 QPS、令牌桶拒绝数、Stream 积压。

有了这套全 Redis 侧的“闸 + 合并 + 负缓存 + 打散 + 广播 + 异步回源” ，即使峰值 10 万 QPS 的查询，也能把“真正落库”的请求稳定压在一个可控水平（例如每秒几百），Mongo 不会被尖峰打穿。

如果要扛1万~10万 QPS 的“查当前换电订单/订单列表” ，但Redis 一旦未命中就会把流量瞬间打爆 MongoDB，做法要从“读模型 + 缓存治理 + 限流与合并 + Mongo 优化”四层一起上。下面给你一套可直接落地的方案与示例代码。

总体思路（先定规则）

  1. 读模型前置到 Redis：把“用户→当前订单ID”“订单详情”维护在 Redis，绝大多数查询不落库。

  2. 两级缓存 + 防击穿：L1 Caffeine（进程内 3060s）+ L2 Redis（510min，带随机抖动）；负缓存、单飞/互斥锁、异步刷新。

  3. 水闸与合并：严格限制每秒能落 Mongo 的请求数（全局/单键），同键请求合并一次查库。

  4. Mongo 只做“回源” ：只在冷启动/失效时查库；建覆盖索引、精确投影，必要时分片或读从。

Redis 键设计（建议）

  • 当前订单映射：ord:cur:{op}:{uid} -> {orderId|NULL}（TTL 120~300s，随机±10% ）

  • 订单详情：ord:det:{orderId} -> JSON（TTL 300~600s，±10%）

  • 热点保护（可选）：复制 N 份 ord:cur:{op}:{uid}#1..N，读随机挑一份。

  • 回源互斥锁：lock:ord:cur:{op}:{uid}（过期 5s）。

  • L1 本地缓存：同样两个 key，TTL 30~60s。

负缓存：当确定“无当前订单”时，把 ord:cur:* 设为 "NULL"，TTL 30~60s，防穿透。

查询流程（防雪崩/防击穿/合并）

以“查用户当前订单”为例：

  1. L1 命中 → 直接返回。

  2. L2 命中（Redis）

  • 值为 "NULL" → 直接返回“无”。

  • 有 orderId → 继续查 ord:det:{orderId}，缺则批量 MGET/Pipeline 获取，仍缺再按互斥逻辑回源。

  • 缓存都未命中 → 尝试获取互斥锁（Redisson tryLock / Redis SETNX）

    • 拿到锁：再检查一次 Redis（双检），仍然未命中 → 限流器允许后回源 Mongo，写回 Redis（含负缓存），发布失效广播清 L1；释放锁。

    • 没拿到锁：短暂自旋/订阅通知（最多 50~100ms），若仍没有则返回旧值/降级结果（避免把并发都压到 Mongo）。

    可选：refresh-ahead（软TTL）——距离过期很近时由后台线程提前刷新；前台仍用旧值，避免尖峰时全部同时失效。

    代码骨架（Spring Boot）

    1) 服务方法（L1 + L2 + 单飞 + 负缓存）

    @Service
public class OrderQueryService {

    private final Cache<String, String> l1 = Caffeine.newBuilder()
            .maximumSize(300_000).expireAfterWrite(Duration.ofSeconds(45)).build();

    @Autowired StringRedisTemplate srt;
    @Autowired RedissonClient redisson;
    @Autowired OrderRepository orderRepo; // Mongo 回源
    @Autowired RateLimiter mongoLimiter; // 每实例/全局限流（如令牌桶）

    private static final String NULL = "NULL";

    public OrderDto getCurrentOrder(String op, String uid) {
        String curKey = "ord:cur:" + op + ":" + uid;

        // L1
        String ordId = l1.getIfPresent(curKey);
        if (ordId != null) return fetchDetailOrNull(ordId);

        // L2
        ordId = srt.opsForValue().get(curKey);
        if (NULL.equals(ordId)) { l1.put(curKey, NULL); return null; }
        if (ordId != null) { 
            l1.put(curKey, ordId);
            return fetchDetailOrNull(ordId);
        }

        // 互斥 + 双检
        RLock lock = redisson.getLock("lock:" + curKey);
        boolean locked = false;
        try {
            locked = lock.tryLock(0, 5, TimeUnit.SECONDS);
            if (locked) {
                // 双检
                ordId = srt.opsForValue().get(curKey);
                if (ordId != null) {
                    l1.put(curKey, ordId);
                    return fetchDetailOrNull(ordId);
                }

                // 限流：保护 Mongo
                if (!mongoLimiter.tryAcquire()) {
                    // 返回降级：可返回空/提示“稍后重试”/旧快照
                    return null;
                }

                // 回源 Mongo（必须有覆盖索引）
                OrderDoc doc = orderRepo.findCurrentByUser(op, uid); // 精确查
                if (doc == null) {
                    setWithJitter(curKey, NULL, 30, 60);
                    l1.put(curKey, NULL);
                    return null;
                }

                ordId = doc.getOrderId();
                // 写回 Redis（当前id + 详情）
                setWithJitter(curKey, ordId, 120, 300);
                srt.opsForValue().set("ord:det:" + ordId, toJson(doc), getJitter(300, 600));

                // 通知各实例清 L1（可用 Redis Pub/Sub）
                srt.convertAndSend("cache:invalidate", curKey);

                l1.put(curKey, ordId);
                return map(doc);
            } else {
                // 未拿到锁：短暂等他人填充
                long end = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(80);
                while (System.nanoTime() < end) {
                    String v = srt.opsForValue().get(curKey);
                    if (v != null) return NULL.equals(v) ? null : fetchDetailOrNull(v);
                    Thread.onSpinWait();
                }
                // 仍无：返回降级结果
                return null;
            }
        } catch (InterruptedException ie) {
            Thread.currentThread().interrupt();
            return null;
        } finally {
            if (locked) lock.unlock();
        }
    }

    private OrderDto fetchDetailOrNull(String ordId) {
        if (NULL.equals(ordId)) return null;
        String detKey = "ord:det:" + ordId;
        String json = l1.getIfPresent(detKey);
        if (json != null) return fromJson(json);

        json = srt.opsForValue().get(detKey);
        if (json != null) { l1.put(detKey, json); return fromJson(json); }

        // 详情缺失：轻回源或延迟刷新（不要高频落库）
        return null; // 或放宽：小概率回源 + 写回
    }

    private void setWithJitter(String key, String val, int minSec, int maxSec) {
        srt.opsForValue().set(key, val, getJitter(minSec, maxSec));
    }
    private Duration getJitter(int min, int max) {
        int ttl = ThreadLocalRandom.current().nextInt(min, max + 1);
        return Duration.ofSeconds(ttl);
    }
}

    上面包含：L1/L2、负缓存、互斥锁、双检、Mongo 限流、短暂自旋合并。
    生产可再加：本地 singleflight（ConcurrentHashMap<String,CompletableFuture<?>>） ，进一步合并同键请求。

    2) 写路径（确保缓存一致）

    • Cache-Aside（推荐） ：订单状态变更/创建 → 先写 Mongo 成功 → 删除/更新 Redis

    • 若是“当前订单”字段：变更时同时更新 ord:cur:* 与 ord:det:*（或直接删，等读端再填）。

    • 广播 L1 失效：convertAndSend("cache:invalidate", key)；各实例订阅后 l1.invalidate(key)。

    保护 Mongo 的“四道闸”

    1. 令牌桶/信号量：单实例/全局控制落库 QPS（例如 1000/s）；超出直接降级。

    2. 每键互斥：同用户同一时刻只有一个回源。

    3. 批量/聚合（列表页）：一次取多条订单 ID → MGET 详情，缺的集中补。

    4. 负缓存 + 软TTL刷新：无数据也缓存，提前刷新避免同刻过期雪崩。

    MongoDB 优化（必须做）

    • 覆盖索引

      • 当前订单：{ operatorId:1, userId:1, status:1 }（partial：status in ['CREATED','PAYING','RUNNING']）

      • 列表分页：{ operatorId:1, userId:1, createdAt:-1 }（再投影需要的字段）

    • 精确投影：只取页面需要的字段，减对象体积。

    • 连接池：maxPoolSize 合理（100~500/实例），waitQueueTimeoutMS 限制排队。

    • 读从（可选） ：允许略旧可用 secondaryPreferred（当前订单通常不建议）。

    • 分片：量级特别大按 {operatorId, userId} 或 {operatorId, createdAt} 分片，避热点。

    额外增强（按需）

    • Bloom Filter / Bitmap：维护“用户是否可能有进行中订单”的布隆过滤器，过滤无效落库。

    • 热点复制：ord:cur:*#1..N 写 N 份，读随机一份；写时用 Lua 一次性更新全部副本。

    • 预热：登录成功、创建/变更订单时顺手把 ord:cur/ord:det 写进缓存。

    • DLQ/监控：指标上报命中率、落库QPS、锁等待、批量大小、Mongo p95/p99。

    一句话总结

    • 读模型进 Redis + L1/L2 两级缓存 → 把 99% 查询挡在缓存；

    • 互斥/合并 + 负缓存 + 随机TTL + 限流 → 防止未命中瞬间把 Mongo 打爆；

    • Mongo 侧覆盖索引 + 精确投影 → 回源也快；

    • 这套下来，10万 QPS 的读流量可稳住在几百到几千 QPS以内落库，系统不抖。

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布