引言

有序集合（Sorted Set，簡稱 ZSet）是一種非常強大且常用的數據結構。它既像集合（Set）一樣保證成員（member）的唯一性，又允許為每個成員關聯一個分數（score），并能根據分數對成員進行高效排序。

一、 ZSet 核心概念與特性

1.1 什么是 ZSet？

ZSet 是 Redis 提供的一種有序集合數據結構。你可以把它想象成一個集合，但這個集合里的每個元素（我們稱之為成員 member）都額外綁定了一個浮點數類型的分數 score。ZSet 最重要的特性就是它會根據 score 對 member 進行排序。

核心特性總結：

1. 成員唯一性 (Member Uniqueness): 和普通的 Set 一樣，ZSet 中的 member 是唯一的，不允許重復。如果你嘗試添加一個已經存在的 member，只會更新它的 score。
2. 分數關聯 (Score Association): 每個 member 都必須關聯一個 score。這個 score 是一個浮點數，可以是正數、負數、零，甚至是 +inf (正無窮) 或 -inf (負無窮)。score 可以重復，即不同的 member 可以有相同的 score。
3. 有序性 (Ordered): ZSet 內部的元素是根據 score 從小到大排序的。如果 score 相同，則按照 member 的字典序 (lexicographical order) 進行排序（具體取決于 Redis 版本和配置，但通常是這樣）。
4. 高效操作: ZSet 提供了一系列高效的操作，比如添加/刪除成員、更新分數、根據分數范圍獲取成員、根據成員獲取排名/分數等。

1.2 ZSet 與 Set、List 的本質區別

為了更好地理解 ZSet，我們將其與 Redis 中另外兩種常用的集合類型 Set 和 List 進行比較：

特性	ZSet (有序集合)	Set (集合)	List (列表)
存儲內容	唯一的 member + 關聯的 score	唯一的 member	member (可重復)
有序性	根據 score 排序 (score 相同按字典序)	無序	按插入順序排序
成員唯一	是	是	否
核心優勢	高效的排序、范圍查找、排名計算	快速判斷成員是否存在、去重	按順序存取、模擬棧/隊列
典型命令	ZADD, ZRANGE, ZRANK, ZRANGEBYSCORE	SADD, SISMEMBER, SMEMBERS	LPUSH, RPOP, LRANGE

總結來說：

• 如果你只需要存儲唯一的值，不關心順序，用 Set。
• 如果你需要按照元素添加的順序存儲，并且允許重復，用 List。
• 如果你需要存儲唯一的元素，并且希望能夠根據某個權重（分數）進行排序和范圍查找，那么 ZSet 是不二之選。

---

二、 ZSet 典型應用場景

ZSet 的有序性和高效范圍查詢能力使其在眾多業務場景中大放異彩。以下是一些常見的應用示例：

2.1 排行榜 (Leaderboards)

這是 ZSet 最經典的應用場景之一。例如：

• 游戲積分排行榜：member 是玩家 ID，score 是玩家積分。
• 用戶貢獻排行榜：member 是用戶 ID，score 是貢獻值（如發帖數、獲贊數）。
• 商品銷量排行榜：member 是商品 ID，score 是銷量或銷售額。

如何實現：

1. 添加/更新用戶積分： 使用 ZADD 命令。如果用戶已存在，則更新其分數；如果不存在，則添加。

   # 添加或更新玩家 Alice 的積分為 1500
   ZADD game:leaderboard 1500 Alice
   # 添加或更新玩家 Bob 的積分為 2000
   ZADD game:leaderboard 2000 Bob
   # 使用 NX 選項，僅當 Bob 不存在時才添加 (這里不會成功，因為 Bob 已存在)
   ZADD game:leaderboard NX 2100 Bob
   # 使用 XX 選項，僅當 Alice 存在時才更新 (會成功)
   ZADD game:leaderboard XX 1600 Alice
   # 使用 INCR 選項，給 Alice 增加 100 分 (變為 1700)
   ZADD game:leaderboard INCR 100 Alice
   

2. 獲取排名前 N 的用戶： 使用 ZREVRANGE (按分數從高到低排序)。

   # 獲取積分榜前 3 名的玩家及其分數
   ZREVRANGE game:leaderboard 0 2 WITHSCORES
   # 輸出可能類似：1) "Bob" 2) "2000" 3) "Alice" 4) "1700"
   

3. 獲取用戶排名： 使用 ZREVRANK (從高到低排名，0 是第一名) 或 ZRANK (從低到高排名)。

   # 獲取 Alice 的排名 (從高到低)
   ZREVRANK game:leaderboard Alice
   # 輸出可能類似：1 (表示第二名，因為 Bob 是 0)
   

4. 獲取用戶分數： 使用 ZSCORE。

   # 獲取 Bob 的分數
   ZSCORE game:leaderboard Bob
   # 輸出: "2000"
   

5. 獲取指定分數范圍內的用戶： 使用 ZRANGEBYSCORE 或 ZREVRANGEBYSCORE。

   # 獲取分數在 1500 到 1800 之間的玩家 (包含邊界)
   ZRANGEBYSCORE game:leaderboard 1500 1800 WITHSCORES
   # 輸出可能類似：1) "Alice" 2) "1700"
   

2.2 帶權重的任務隊列 / 延遲隊列

有時我們需要處理一些帶有優先級的任務，或者需要在未來的某個特定時間點執行的任務。ZSet 可以很好地模擬這種場景。

• member: 任務的唯一標識符 (如任務 ID)。
• score:
  • 優先級: 分數越小（或越大，取決于業務定義）表示優先級越高。
  • 執行時間: 分數存儲任務應該被執行的 Unix 時間戳。

如何實現：

1. 添加任務：

   # 添加一個優先級為 10 (數值越小優先級越高) 的任務 task:1
   ZADD priority_queue 10 task:1
   # 添加一個需要在未來時間戳 1678886400 執行的任務 task:2
   ZADD delayed_queue 1678886400 task:2
   

2. 獲取優先級最高的任務： 使用 ZRANGE (取出分數最低的) 或 ZREVRANGE (取出分數最高的)。通常我們會結合 LIMIT 取出若干個。

   # 取出優先級最高的 1 個任務 (分數最低)
   ZRANGE priority_queue 0 0
   # 輸出: 1) "task:1"
   

3. 獲取到期需要執行的任務： 使用 ZRANGEBYSCORE 獲取所有 score 小于等于當前時間戳的任務。

   # 假設當前時間戳是 1678886405
   # 取出所有 score 在 -inf 到 1678886405 之間的任務
   ZRANGEBYSCORE delayed_queue -inf 1678886405 WITHSCORES
   # 輸出可能類似: 1) "task:2" 2) "1678886400"
   

4. 處理并移除任務： 獲取到任務后，需要使用 ZREM 將其從隊列中移除，防止重復處理。這是一個關鍵步驟，為了保證原子性（獲取并刪除），通常會結合 Lua 腳本來實現。

   -- Lua 腳本示例：原子地獲取并刪除優先級最高的任務
   local key = KEYS[1]
   local tasks = redis.call('ZRANGE', key, 0, 0) -- 獲取分數最低的任務
   if #tasks > 0 thenredis.call('ZREM', key, tasks[1]) -- 如果存在則刪除return tasks[1] -- 返回任務 ID
   elsereturn nil -- 沒有任務
   end
   

   或者使用 Redis 5.0+ 提供的阻塞彈出命令 BZPOPMIN 或 BZPOPMAX，它們能原子地完成“獲取并刪除”操作，并且在隊列為空時阻塞等待。

   # 阻塞等待，直到獲取并刪除 delayed_queue 中分數最低的任務，超時時間 60 秒
   BZPOPMIN delayed_queue 60
   # 輸出類似: 1) "delayed_queue" 2) "task:2" 3) "1678886400.0"
   

2.3 時間軸 (Timeline)

在社交應用中，用戶的 Feed 流（時間軸）通常需要展示關注的人發布的最新內容。

• member: 內容的唯一 ID (如帖子 ID)。
• score: 內容的發布時間戳。

如何實現：

1. 發布內容時添加到關注者的 Timeline ZSet：

   # 用戶 UserA 發布了帖子 PostX (時間戳 1678887000)
   # 將 PostX 添加到關注者 Follower1 和 Follower2 的 Timeline ZSet
   ZADD timeline:Follower1 1678887000 PostX
   ZADD timeline:Follower2 1678887000 PostX
   

2. 用戶拉取最新內容： 使用 ZREVRANGE 按時間戳倒序獲取最新的 N 條內容 ID。

   # Follower1 拉取最新的 10 條內容 ID
   ZREVRANGE timeline:Follower1 0 9
   

   拿到內容 ID 后，再根據 ID 去獲取內容的具體信息（通常存儲在 Hash 或 String 中）。

2.4 范圍查找

根據某個數值范圍進行查找，例如：

• 查找價格在 100 到 200 元之間的商品。
• 查找年齡在 18 到 25 歲之間的用戶。
• 查找距離某個點特定范圍內的地點（結合 GeoHash）。

如何實現：

使用 ZRANGEBYSCORE 或 ZREVRANGEBYSCORE。

# 假設有一個 ZSet 存儲商品價格: member 是商品 ID, score 是價格
ZADD product_prices 150 product:1 99.9 product:2 210 product:3 180 product:4# 查找價格在 100 到 200 (包含邊界) 之間的商品
ZRANGEBYSCORE product_prices 100 200 WITHSCORES
# 輸出: 1) "product:1" 2) "150" 3) "product:4" 4) "180"# 查找價格在 (100, 200] 之間的商品 (不包含 100)
ZRANGEBYSCORE product_prices (100 200 WITHSCORES

---

三、 ZSet 底層實現

為了實現高效的排序和查找，Redis ZSet 的底層實現會根據存儲數據的規模動態選擇不同的編碼方式。主要有兩種：ziplist 和 skiplist + dict。

3.1 ziplist (壓縮列表)

觸發條件：

當 ZSet 同時滿足以下兩個條件時，會優先采用 ziplist 編碼：

1. ZSet 中元素的數量小于 zset_max_ziplist_entries 配置的值（默認 128）。
2. ZSet 中每個元素（member 和 score）的字節長度都小于 zset_max_ziplist_value 配置的值（默認 64）。

結構與原理：

ziplist 是一種設計非常緊湊的連續內存數據結構，旨在盡可能地節省內存。它不像普通的數組那樣每個元素占用固定大小的空間，而是根據實際內容動態調整每個節點的長度。

一個 ziplist 的大致結構如下：

<zlbytes> <zltail> <zllen> <entry1> <entry2> ... <entryN> <zlend>

• zlbytes: (4 字節) 記錄整個 ziplist 占用的總字節數。
• zltail: (4 字節) 記錄到最后一個 entry 的偏移量，用于快速定位表尾。
• zllen: (2 字節) 記錄 ziplist 中的 entry 數量。當數量超過 65535 時，該字段固定為 65535，需要遍歷才能確定實際數量。
• entryX: 實際存儲數據的節點。每個 entry 包含前一個 entry 的長度信息（用于反向遍歷）和當前 entry 的編碼及內容。member 和 score 在 ziplist 中是緊鄰存儲的兩個 entry。
• zlend: (1 字節) 特殊標記，值為 0xFF，表示 ziplist 的末尾。

entry 的結構（重點）：

<prevrawlen> <encoding> <content>

• prevrawlen: 記錄前一個 entry 的總長度。這個字段的長度本身是可變的（1字節或5字節），用于支持從后向前遍歷 ziplist。
• encoding: 記錄當前 content 的編碼方式（字符串還是整數）以及長度。
• content: 實際存儲的數據（member 或 score）。score 會被存儲為字符串形式。

ziplist 如何存儲 ZSet 元素：

在 ziplist 中，一個 ZSet 元素由兩個相鄰的 entry 表示：第一個 entry 存儲 member，第二個 entry 存儲 score。它們是成對出現的。

源碼片段 (ziplist.c 附近，概念性展示，非精確代碼):

// 概念性展示 ziplist entry 結構
typedef struct zlentry {unsigned int prevrawlensize; // 存儲前一個節點長度所需的字節數 (1 或 5)unsigned int prevrawlen;     // 前一個節點的長度unsigned int lensize;        // 存儲當前節點 content 長度或類型所需的字節數unsigned int len;            // 當前節點 content 的長度unsigned int headersize;     // 當前節點頭部 (prevrawlen + encoding) 的總大小unsigned char encoding;      // 編碼類型unsigned char *p;            // 指向當前節點數據的指針 (content)
} zlentry;// 在 ziplist 中查找元素大致需要遍歷比較
// 插入或刪除可能引起連鎖更新

優點：

• 內存效率高: 連續存儲，沒有指針開銷（相比鏈表），節點長度可變，非常節省內存。

缺點：

• 查找效率較低: 查找特定 member 或 score 需要遍歷 ziplist，時間復雜度為 O(N)，N 是元素數量。范圍查找效率也不高。
• 連鎖更新 (Cascade Update): 這是 ziplist 最大的問題。當插入或刪除一個 entry 時，如果導致后續 entry 的 prevrawlen 字段長度發生變化（比如從 1 字節變成 5 字節），就可能需要調整后續所有 entry 的位置，引發連鎖反應，導致操作的時間復雜度在最壞情況下變為 O(N^2)。更新操作也可能觸發。

3.2 skiplist (跳躍表) + dict (哈希表)

觸發條件：

當 ZSet 不再滿足 ziplist 的編碼條件時（即元素數量超過 zset_max_ziplist_entries 或任一元素的長度超過 zset_max_ziplist_value），Redis 會自動將其編碼轉換為 skiplist + dict。注意：這個轉換是單向的，一旦變成 skiplist，即使后來元素減少，也不會自動轉回 ziplist。

結構與原理：

這種編碼方式結合了跳躍表 (skiplist) 和哈希表 (dict) 的優點：

• dict (哈希表): 用于存儲從 member 到 score 的映射。這使得通過 member 快速查找其對應的 score (如 ZSCORE 命令) 的平均時間復雜度達到 O(1)。
• skiplist (跳躍表): 用于存儲所有 ZSet 元素，并按照 score 進行排序。跳躍表是一種通過多層有序鏈表實現高效查找、插入、刪除的數據結構，其操作的平均時間復雜度為 O(log N)，最壞情況下為 O(N)。它特別擅長進行范圍查找（如 ZRANGE, ZRANGEBYSCORE）。

為什么需要兩者結合？

• 如果只用 dict，無法高效地按 score 排序和范圍查找。
• 如果只用 skiplist，雖然也能通過 member 找到 score（遍歷 skiplist），但平均時間復雜度是 O(log N)，不如 dict 的 O(1) 高效。ZSCORE 是常用操作，效率很重要。

兩者結合，可以在 O(1) 時間內通過 member 獲取 score，同時在 O(log N) 時間內完成基于 score 的排序、排名和范圍查找。

skiplist (跳躍表) 詳解 (重點和難點)：

跳躍表是一種概率性數據結構，它在有序鏈表的基礎上增加了額外的“快速通道”（前向指針），從而實現類似二分查找的效率。

結構：

一個跳躍表包含：

• header: 頭節點，不存儲實際數據，但包含指向各層鏈表頭部的指針。
• tail: 指向跳躍表尾節點的指針。
• length: 跳躍表中節點的數量。
• level: 跳躍表中當前最高的層數。

每個跳躍表節點 (zskiplistNode) 包含：

• member (ele): ZSet 的成員。
• score: ZSet 的分數。
• backward: 指向前一個節點的指針（用于反向遍歷，如 ZREVRANGE）。
• level[]: 一個柔性數組（flexible array member），存儲該節點在每一層的前向指針 (forward) 和跨度 (span)。
  • forward: 指向該層鏈表中下一個節點的指針。
  • span: 表示當前節點的 forward 指針指向的節點與當前節點之間跨越了多少個節點。span 對于快速計算排名 (ZRANK) 至關重要。

圖解 skiplist：

          header                                                     tail
level 3:   head ->-------------------------------------------------> NULL
level 2:   head ->------------------------> node E -----------------> NULL
level 1:   head ->--------> node B ------> node E ->-----> node G --> NULL
level 0:   head --> node A -> node B -> node C -> node E -> node F -> node G -> NULL
(score)            (10)      (20)      (30)      (50)      (60)      (70)(span)
level 3:            span=7
level 2:            span=4                   span=3
level 1:            span=2        span=2               span=2
level 0:            span=1        span=1     span=1    span=1     span=1     span=1

• 第 0 層包含所有節點，是一個標準的有序鏈表。
• 更高層級的鏈表是第 0 層的“快速通道”，它們只包含部分節點。
• 一個節點會出現在多少層（除了第 0 層）是隨機決定的，但層數越高的概率越低（通常是 P=1/4 或 1/2）。這種隨機性保證了跳躍表在插入、刪除、查找操作上的平均時間復雜度為 O(log N)。
• span 表示從當前節點沿著該層的 forward 指針跳到下一個節點，中間跳過了多少個底層節點。例如，在 level 2，header 的 forward 指向 node E，span 為 4，表示從 header 到 node E 之間有 4 個節點 (A, B, C, E)。

查找過程 (例如查找 score=50 的節點 E):

1. 從最高層 (level 3) 的 header 開始。forward 指向 NULL，比 50 大。
2. 下降到 level 2。header 的 forward 指向 node E (score 50)。找到目標。

查找過程 (例如查找 score=65 的節點，它不存在):

1. 從最高層 (level 3) 的 header 開始。forward 指向 NULL，比 65 大。
2. 下降到 level 2。header 的 forward 指向 node E (50)，比 65 小。沿著 level 2 的 forward 到達 node E。node E 在 level 2 的 forward 指向 NULL，比 65 大。
3. 下降到 level 1。node E 的 forward 指向 node G (70)，比 65 大。
4. 下降到 level 0。node E 的 forward 指向 node F (60)，比 65 小。沿著 level 0 的 forward 到達 node F。node F 的 forward 指向 node G (70)，比 65 大。
5. 此時，我們位于 node F，并且下一節點 score (70) 大于目標 (65)。查找結束，目標不存在。插入位置應該在 F 和 G 之間。

計算排名 (ZRANK，例如計算 node E 的排名):

1. 從最高層開始查找，累加跨過的 span 值。
2. Level 3: header -> NULL。不前進。rank=0。
3. Level 2: header -> node E。前進了，累加 header 在 level 2 的 span (假設為 4)。rank = 4。找到目標。
4. 排名是 rank - 1 (因為排名從 0 開始)，所以 node E 的排名是 3。

源碼片段 (server.h, t_zset.c 附近，結構定義和關鍵操作)：

/* server.h */
#define ZSKIPLIST_MAXLEVEL 32 /* 跳躍表最大層數 */
#define ZSKIPLIST_P 0.25      /* 用于計算隨機層數的概率 *//* 跳躍表節點 */
typedef struct zskiplistNode {sds ele; // 成員 (member)double score; // 分數struct zskiplistNode *backward; // 后向指針// 層級數組，包含前向指針和跨度struct zskiplistLevel {struct zskiplistNode *forward; // 前向指針unsigned long span; // 跨度} level[]; // 柔性數組
} zskiplistNode;/* 跳躍表 */
typedef struct zskiplist {struct zskiplistNode *header, *tail; // 頭尾節點指針unsigned long length; // 節點數量int level; // 當前最大層數
} zskiplist;/* ZSet 結構 */
typedef struct zset {dict *dict; // 哈希表，member -> scorezskiplist *zsl; // 跳躍表，按 score 排序
} zset;/* t_zset.c - zslInsert: 插入新節點到跳躍表 */
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {zskiplistNode *update[ZSKIPLIST_MAXLEVEL]; // 記錄每層查找路徑上需要更新 forward 指針的節點unsigned int rank[ZSKIPLIST_MAXLEVEL]; // 記錄每層查找到的位置的排名 (用于計算 span)zskiplistNode *x;int i, level;// 1. 查找插入位置，并記錄路徑 (update 數組) 和排名 (rank 數組)x = zsl->header;for (i = zsl->level-1; i >= 0; i--) {rank[i] = (i == zsl->level-1) ? 0 : rank[i+1]; // 初始化排名// 在當前層向右查找，直到找到第一個 score 更大或 ele 字典序更大的節點while (x->level[i].forward &&(x->level[i].forward->score < score ||(x->level[i].forward->score == score &&sdscmp(x->level[i].forward->ele,ele) < 0))) // 如果分數相同，比較成員字典序{rank[i] += x->level[i].span; // 累加跨度到排名x = x->level[i].forward; // 前進}update[i] = x; // 記錄該層需要更新 forward 指針的節點}// 2. 計算新節點的隨機層數level = zslRandomLevel(); // 根據概率 P 計算一個隨機層數// 3. 如果新層數高于當前跳躍表最大層數，初始化 update 和 rank 數組中新層的數據if (level > zsl->level) {for (i = zsl->level; i < level; i++) {rank[i] = 0;update[i] = zsl->header;update[i]->level[i].span = zsl->length; // 新層的 header span 為當前總長度}zsl->level = level; // 更新跳躍表最大層數}// 4. 創建新節點x = zslCreateNode(level,score,ele); // 分配內存并初始化新節點// 5. 更新每一層的 forward 指針和 spanfor (i = 0; i < level; i++) {// 將新節點插入到 update[i] 和原 update[i]->level[i].forward 之間x->level[i].forward = update[i]->level[i].forward;update[i]->level[i].forward = x;// 更新 span 值// 新節點的 span = 原 update[i] 的 span - (新節點之前經過的節點數 rank[0] - update[i] 之前的節點數 rank[i])x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);// update[i] 的 span = (新節點之前經過的節點數 rank[0] - update[i] 之前的節點數 rank[i]) + 1 (新節點本身)update[i]->level[i].span = (rank[0] - rank[i]) + 1;}// 6. 如果新節點的層數低于原跳躍表最大層數，更新未涉及層級的 span (加 1，因為多了一個節點)for (i = level; i < zsl->level; i++) {update[i]->level[i].span++;}// 7. 更新新節點的 backward 指針x->backward = (update[0] == zsl->header) ? NULL : update[0];if (x->level[0].forward)x->level[0].forward->backward = x; // 更新后繼節點的 backward 指針elsezsl->tail = x; // 如果新節點是最后一個節點，更新 tail 指針// 8. 更新跳躍表長度zsl->length++;return x;
}

優點：

• 高效的查找、插入、刪除: 平均時間復雜度為 O(log N)。
• 高效的范圍查詢: 按 score 范圍查找非常快。
• 高效的排名計算: 利用 span 可以在 O(log N) 內計算排名。

缺點：

• 內存開銷相對較大: 相較于 ziplist，需要存儲額外的指針 (forward, backward) 和 span 信息，內存占用更多。

3.3 編碼轉換

• ziplist -> skiplist: 當 ZSet 不再滿足 ziplist 的兩個條件（元素數量或大小超限）時，Redis 會自動執行轉換。這個過程需要創建新的 dict 和 skiplist，并將 ziplist 中的所有元素逐一添加到新的數據結構中。這是一個一次性的、相對耗時的操作，會消耗額外的 CPU 和內存。但轉換完成后，后續操作將受益于 skiplist 的高效率。
• skiplist -> ziplist: Redis 不會自動執行此轉換。即使 ZSet 的元素數量和大小降回 ziplist 的閾值以下，它仍然會保持 skiplist 編碼。

你可以使用 OBJECT ENCODING key 命令查看一個 ZSet 當前使用的底層編碼。

---

四、 ZSet 常用命令詳解

下面我們詳細介紹 ZSet 的常用命令，包括其功能、參數、返回值、時間復雜度和示例。時間復雜度會根據底層編碼（ziplist 或 skiplist）有所不同。

復雜度說明：

• N: ZSet 中的元素數量。
• M: 被操作的元素數量。
• LogN: 通常指 log base 2 of N。

4.1 添加與更新

• ZADD key [NX|XX] [CH] [INCR] score member [score member ...]

  • 功能: 向有序集合添加一個或多個成員，或者更新已存在成員的分數。
  • 參數:
    • NX: 僅當成員不存在時才添加。不更新已存在的成員。
    • XX: 僅當成員存在時才更新。不添加新成員。
    • CH: (Changed) 返回值從“新添加成員的數量”變為“被修改成員的總數”（包括添加和更新）。
    • INCR: 將命令模式從“添加/更新”變為“增加分數”。此時只能指定一對 score member，score 表示要增加的值（可以是負數）。如果 member 不存在，則添加它，score 為指定的增量值。
  • 返回值: 默認情況下，返回新添加到集合中的成員數量（不包括分數被更新的成員）。如果使用了 CH 選項，返回被修改（添加或更新）的成員總數。如果使用了 INCR 選項，返回成員的新分數（字符串形式）。
  • 復雜度:
    • 添加單個元素：O(log N) (skiplist) / 平均 O(log N)，最壞 O(N^2) 因連鎖更新 (ziplist)
    • 添加多個元素：O(M * log N) (skiplist) / 平均 O(M * log N)，最壞 O(N*M) 或 O(N^2) (ziplist)
  • 示例: (見應用場景部分)

• ZINCRBY key increment member

  • 功能: 為有序集合中指定成員的分數增加 increment。如果成員不存在，則添加它，分數等于 increment。相當于 ZADD key INCR increment member。
  • 參數:
    • increment: 要增加的分數值（浮點數）。
    • member: 要操作的成員。
  • 返回值: 成員的新分數（字符串形式）。
  • 復雜度: O(log N) (skiplist) / 平均 O(log N)，最壞 O(N^2) (ziplist)
  • 示例:

    ZADD scores 10 user1
    ZINCRBY scores 5 user1  # user1 的分數變為 15
    # 返回: "15"
    ZINCRBY scores 3 user_new # user_new 不存在，添加，分數為 3
    # 返回: "3"
    

4.2 刪除

• ZREM key member [member ...]

  • 功能: 移除有序集合中的一個或多個成員。忽略不存在的成員。
  • 返回值: 被成功移除的成員數量。
  • 復雜度:
    • 移除單個元素：O(log N) (skiplist) / 平均 O(log N)，最壞 O(N^2) (ziplist)
    • 移除多個元素：O(M * log N) (skiplist) / 平均 O(M * log N)，最壞 O(N*M) 或 O(N^2) (ziplist)
  • 示例:

    ZADD myzset 1 one 2 two 3 three
    ZREM myzset one four # 移除 one 和 four (four 不存在)
    # 返回: 1
    

• ZREMRANGEBYRANK key start stop

  • 功能: 移除有序集合中指定排名范圍內的所有成員。排名按分數從小到大計算，0 是第一個，-1 是最后一個。
  • 參數:
    • start, stop: 排名范圍（包含邊界）。
  • 返回值: 被移除成員的數量。
  • 復雜度: O(log N + M) (skiplist) / O(N) (ziplist)，M 是被移除的數量。
  • 示例:

    ZADD myzset 1 a 2 b 3 c 4 d 5 e
    # 移除排名 0 到 1 的成員 (a, b)
    ZREMRANGEBYRANK myzset 0 1
    # 返回: 2
    # 剩余: c, d, e
    # 移除排名最后 2 位的成員 (d, e)
    ZREMRANGEBYRANK myzset -2 -1
    # 返回: 2
    # 剩余: c
    

• ZREMRANGEBYSCORE key min max

  • 功能: 移除有序集合中指定分數范圍內的所有成員。
  • 參數:
    • min, max: 分數范圍。默認包含邊界。可以使用 ( 開頭表示不包含最小值/最大值，如 (10 20 表示 > 10 且 <= 20。可以使用 -inf 和 +inf 表示負無窮和正無窮。
  • 返回值: 被移除成員的數量。
  • 復雜度: O(log N + M) (skiplist) / O(N) (ziplist)，M 是被移除的數量。
  • 示例:

    ZADD myzset 10 a 20 b 30 c 40 d 50 e
    # 移除分數在 [20, 40] 之間的成員 (b, c, d)
    ZREMRANGEBYSCORE myzset 20 40
    # 返回: 3
    # 剩余: a, e
    # 移除分數 > 45 的成員 (e)
    ZREMRANGEBYSCORE myzset (45 +inf
    # 返回: 1
    # 剩余: a
    

4.3 查詢

• ZCARD key

  • 功能: 獲取有序集合的成員數量（基數）。
  • 返回值: 成員數量。
  • 復雜度: O(1)。無論哪種編碼，長度信息都是直接可用的。
  • 示例:

    ZADD myzset 1 a 2 b
    ZCARD myzset
    # 返回: 2
    

• ZSCORE key member

  • 功能: 獲取指定成員的分數。
  • 返回值: 成員的分數（字符串形式）。如果成員不存在，返回 nil。
  • 復雜度: O(1) (skiplist，通過 dict) / O(N) (ziplist，需要遍歷)。
  • 示例:

    ZADD myzset 1 a
    ZSCORE myzset a
    # 返回: "1"
    ZSCORE myzset non_exist
    # 返回: nil
    

• ZRANK key member

  • 功能: 獲取指定成員的排名（按分數從小到大排序）。排名從 0 開始。
  • 返回值: 成員的排名（整數）。如果成員不存在，返回 nil。
  • 復雜度: O(log N) (skiplist，利用 span) / O(N) (ziplist，需要遍歷)。
  • 示例:

    ZADD myzset 10 a 20 b 30 c
    ZRANK myzset b
    # 返回: 1 (因為 a 是 0, b 是 1)
    

• ZREVRANK key member

  • 功能: 獲取指定成員的排名（按分數從高到低排序）。排名從 0 開始。
  • 返回值: 成員的排名（整數）。如果成員不存在，返回 nil。
  • 復雜度: O(log N) (skiplist) / O(N) (ziplist)。
  • 示例:

    ZADD myzset 10 a 20 b 30 c
    ZREVRANK myzset b
    # 返回: 1 (因為 c 是 0, b 是 1)
    

• ZCOUNT key min max

  • 功能: 獲取有序集合中，分數在指定范圍內的成員數量。
  • 參數:
    • min, max: 分數范圍，語法同 ZREMRANGEBYSCORE。
  • 返回值: 指定分數范圍內的成員數量。
  • 復雜度: O(log N) (skiplist，定位到范圍起點即可) / O(N) (ziplist)。
  • 示例:

    ZADD myzset 10 a 20 b 30 c 40 d 50 e
    ZCOUNT myzset 20 40
    # 返回: 3 (b, c, d)
    ZCOUNT myzset (20 +inf
    # 返回: 3 (c, d, e)
    

• ZRANGE key start stop [WITHSCORES]

  • 功能: 獲取有序集合中指定排名范圍內的成員（按分數從小到大排序）。
  • 參數:
    • start, stop: 排名范圍（包含邊界），0 是第一個，-1 是最后一個，-2 是倒數第二個，以此類推。
    • WITHSCORES: 可選，同時返回成員的分數。
  • 返回值: 成員列表。如果使用 WITHSCORES，則返回 [member1, score1, member2, score2, ...] 格式的列表。
  • 復雜度: O(log N + M) (skiplist) / O(N) (ziplist)，M 是返回的數量。
  • 示例:

    ZADD myzset 10 a 20 b 30 c 40 d
    # 獲取排名 0 到 1 的成員 (a, b)
    ZRANGE myzset 0 1
    # 返回: 1) "a" 2) "b"
    # 獲取排名 1 到 -1 (最后一個) 的成員及其分數 (b, c, d)
    ZRANGE myzset 1 -1 WITHSCORES
    # 返回: 1) "b" 2) "20" 3) "c" 4) "30" 5) "d" 6) "40"
    

• ZREVRANGE key start stop [WITHSCORES]

  • 功能: 獲取有序集合中指定排名范圍內的成員（按分數從高到低排序）。
  • 參數: 同 ZRANGE。
  • 返回值: 同 ZRANGE。
  • 復雜度: O(log N + M) (skiplist) / O(N) (ziplist)。
  • 示例:

    ZADD myzset 10 a 20 b 30 c 40 d
    # 獲取排名 0 到 1 的成員 (按分數從高到低，即 d, c)
    ZREVRANGE myzset 0 1
    # 返回: 1) "d" 2) "c"
    # 獲取所有成員及其分數 (按分數從高到低)
    ZREVRANGE myzset 0 -1 WITHSCORES
    # 返回: 1) "d" 2) "40" 3) "c" 4) "30" 5) "b" 6) "20" 7) "a" 8) "10"
    

• ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count]

  • 功能: 獲取有序集合中，分數在指定范圍內的成員（按分數從小到大排序）。
  • 參數:
    • min, max: 分數范圍，語法同 ZREMRANGEBYSCORE。
    • WITHSCORES: 可選，同時返回成員的分數。
    • LIMIT offset count: 可選，用于分頁。從符合條件的成員中，跳過 offset 個，取出 count 個。
  • 返回值: 成員列表（可能包含分數）。
  • 復雜度: O(log N + M) (skiplist) / O(N) (ziplist)，M 是返回的數量（在 LIMIT 前計算）。
  • 示例:

    ZADD myzset 10 a 20 b 30 c 40 d 50 e
    # 獲取分數在 [20, 40] 之間的成員及其分數 (b, c, d)
    ZRANGEBYSCORE myzset 20 40 WITHSCORES
    # 返回: 1) "b" 2) "20" 3) "c" 4) "30" 5) "d" 6) "40"
    # 獲取分數 > 25 的成員，跳過 1 個，取 2 個 (d, e)
    ZRANGEBYSCORE myzset (25 +inf WITHSCORES LIMIT 1 2
    # 返回: 1) "d" 2) "40" 3) "e" 4) "50"
    

• ZREVRANGEBYSCORE key max min [WITHSCORES] [LIMIT offset count]

  • 功能: 獲取有序集合中，分數在指定范圍內的成員（按分數從高到低排序）。注意 max 和 min 的順序。
  • 參數: 同 ZRANGEBYSCORE，但 max 在前，min 在后。
  • 返回值: 成員列表（可能包含分數）。
  • 復雜度: O(log N + M) (skiplist) / O(N) (ziplist)。
  • 示例:

    ZADD myzset 10 a 20 b 30 c 40 d 50 e
    # 獲取分數在 [40, 20] 之間的成員及其分數 (按分數從高到低，即 d, c, b)
    ZREVRANGEBYSCORE myzset 40 20 WITHSCORES
    # 返回: 1) "d" 2) "40" 3) "c" 4) "30" 5) "b" 6) "20"
    # 獲取分數 <= 35 的成員，按分數從高到低，取前 2 個 (c, b)
    ZREVRANGEBYSCORE myzset 35 -inf WITHSCORES LIMIT 0 2
    # 返回: 1) "c" 2) "30" 3) "b" 4) "20"
    

• ZPOPMIN key [count] (Redis 5.0+)

  • 功能: 移除并返回有序集合中分數最低的一個或多個成員。
  • 參數:
    • count: 可選，指定要移除并返回的成員數量，默認為 1。
  • 返回值: 被移除的成員及其分數列表 [member1, score1, member2, score2, ...]。如果集合為空，返回空列表。
  • 復雜度: O(log N * M)，M 是 count 值。
  • 示例:

    ZADD myzset 10 a 20 b 30 c
    ZPOPMIN myzset 2 # 移除并返回 a 和 b
    # 返回: 1) "a" 2) "10" 3) "b" 4) "20"
    # 集合剩余: c
    

• ZPOPMAX key [count] (Redis 5.0+)

  • 功能: 移除并返回有序集合中分數最高的一個或多個成員。
  • 參數: 同 ZPOPMIN。
  • 返回值: 同 ZPOPMIN。
  • 復雜度: O(log N * M)。
  • 示例:

    ZADD myzset 10 a 20 b 30 c
    ZPOPMAX myzset 1 # 移除并返回 c
    # 返回: 1) "c" 2) "30"
    # 集合剩余: a, b
    

• BZPOPMIN key [key ...] timeout (Redis 5.0+)

  • 功能: ZPOPMIN 的阻塞版本。如果所有指定的 ZSet 都為空，連接將阻塞 timeout 秒，直到有元素可彈出或超時。timeout 為 0 表示無限期阻塞。它會從第一個非空 ZSet 中彈出元素。
  • 返回值: 一個包含 3 個元素的列表：彈出元素的來源 ZSet 的鍵名、被彈出的成員、成員的分數。如果超時，返回 nil。
  • 復雜度: O(log N)。
  • 示例: (用于實現可靠的任務隊列)

    # 阻塞等待從 queue1 或 queue2 中彈出分數最低的元素，最多等 10 秒
    BZPOPMIN queue1 queue2 10
    

• BZPOPMAX key [key ...] timeout (Redis 5.0+)

  • 功能: ZPOPMAX 的阻塞版本。
  • 返回值: 同 BZPOPMIN。
  • 復雜度: O(log N)。

• ZSCAN key cursor [MATCH pattern] [COUNT count]

  • 功能: 迭代有序集合中的元素（成員和分數）。用于遍歷大型 ZSet 而不阻塞服務器。
  • 參數:
    • cursor: 游標，第一次迭代從 0 開始，后續迭代使用上次返回的游標。
    • MATCH pattern: 可選，只返回匹配給定模式的成員。
    • COUNT count: 可選，提示每次迭代返回的元素數量（不保證精確）。
  • 返回值: 一個包含兩個元素的列表：第一個是下一次迭代使用的 cursor（如果返回 “0” 表示迭代完成），第二個是本次迭代返回的元素列表 [member1, score1, member2, score2, ...]。
  • 復雜度: 每次調用 O(M)，M 是 COUNT 值。完整遍歷需要 O(N)。
  • 示例:

    # 第一次迭代
    ZSCAN myzset 0 MATCH user:* COUNT 10
    # 返回: 1) "17"  (下次的 cursor)
    #      2) 1) "user:1" 2) "100" 3) "user:3" 4) "150" ... (最多 10 對)# 后續迭代
    ZSCAN myzset 17 MATCH user:* COUNT 10
    

4.4 集合運算 (交集與并集)

• ZINTERSTORE destination numkeys key [key ...] [WEIGHTS weight [weight ...]] [AGGREGATE SUM|MIN|MAX]

  • 功能: 計算一個或多個有序集合的交集，并將結果存儲在 destination 鍵中。對于交集中的成員，其分數可以按指定方式聚合。
  • 參數:
    • destination: 存儲結果的鍵名。如果已存在，會被覆蓋。
    • numkeys: 要計算交集的 ZSet 數量。
    • key [key ...]: 要計算交集的 ZSet 的鍵名。
    • WEIGHTS weight [weight ...]: 可選，為每個輸入 ZSet 指定一個權重因子，計算交集成員分數時會先乘以權重。默認權重為 1。
    • AGGREGATE SUM|MIN|MAX: 可選，指定如何聚合交集成員的分數。SUM (默認): 各 ZSet 中分數之和 (乘以權重后)。MIN: 取最小值。MAX: 取最大值。
  • 返回值: 存儲在 destination 中的結果集合的成員數量。
  • 復雜度: O(NKLogM) 最壞情況，N 是最小輸入 ZSet 的大小，K 是輸入 ZSet 的數量，M 是結果 ZSet 的大小。通常取決于最小 ZSet 的大小。
  • 示例:

    ZADD zset1 1 one 2 two
    ZADD zset2 1 one 2 two 3 three
    # 計算 zset1 和 zset2 的交集，分數相加，存入 zset_inter
    ZINTERSTORE zset_inter 2 zset1 zset2 WEIGHTS 1 1 AGGREGATE SUM
    # 返回: 2 (交集有 one, two)
    ZRANGE zset_inter 0 -1 WITHSCORES
    # 返回: 1) "one" 2) "2" (1*1 + 1*1) 3) "two" 4) "4" (2*1 + 2*1)ZADD zset3 5 one 1 two 10 four
    # 計算 zset1 和 zset3 交集，zset1 權重 2，zset3 權重 3，取最大分數
    ZINTERSTORE zset_inter2 2 zset1 zset3 WEIGHTS 2 3 AGGREGATE MAX
    # 返回: 2 (交集有 one, two)
    ZRANGE zset_inter2 0 -1 WITHSCORES
    # 返回: 1) "one" 2) "15" (max(1*2, 5*3)) 2) "two" 4) "4" (max(2*2, 1*3))
    

• ZUNIONSTORE destination numkeys key [key ...] [WEIGHTS weight [weight ...]] [AGGREGATE SUM|MIN|MAX]

  • 功能: 計算一個或多個有序集合的并集，并將結果存儲在 destination 鍵中。
  • 參數: 同 ZINTERSTORE。
  • 返回值: 存儲在 destination 中的結果集合的成員數量。
  • 復雜度: O(NLogN + MLogM)，N 是所有輸入 ZSet 成員總數，M 是結果 ZSet 大小。通常取決于所有輸入 ZSet 的總大小。
  • 示例:

    ZADD zset1 1 one 2 two
    ZADD zset2 1 one 2 two 3 three
    # 計算 zset1 和 zset2 的并集，分數相加，存入 zset_union
    ZUNIONSTORE zset_union 2 zset1 zset2 AGGREGATE SUM
    # 返回: 3 (并集有 one, two, three)
    ZRANGE zset_union 0 -1 WITHSCORES
    # 返回: 1) "one" 2) "2" (1+1) 3) "three" 4) "3" 5) "two" 6) "4" (2+2)
    

---

五、 總結

Redis ZSet 是一種功能強大的有序集合數據結構，它通過將唯一的成員與浮點數分數相關聯，并根據分數進行排序，為許多業務場景提供了高效的解決方案。

• 核心優勢: 成員唯一、按 score 排序、高效的排名計算和范圍查找。
• 應用廣泛: 排行榜、延遲隊列、時間軸、地理位置搜索（底層）、范圍查詢等。
• 底層實現: 根據數據規模動態選擇 ziplist（節省內存，小規模數據）或 skiplist + dict（查詢效率高，大規模數據）。skiplist 通過多層鏈表和 span 實現了 O(log N) 的高效操作。
• 命令豐富: 提供了 ZADD, ZREM, ZRANGE, ZRANK, ZRANGEBYSCORE 等一系列覆蓋增刪改查、范圍查詢、集合運算的命令。