本文將探討 Redis 的 List, Set, Sorted Set 數據結構，通過 Python 代碼示例展示如何構建消息隊列、實現社交關係（共同好友）和實時排行榜，解決高併發下的常見業務難題。

前言

在掌握了 String 和 Hash 後，我們將探索 Redis 更為強大的三種數據結構：**List（列表）**、**Set（集合）** 和 **Sorted Set（有序集合）**。它們是構建複雜功能的基石，能優雅地解決消息隊列、社交關係、實時排行榜等高級場景問題。

​本篇讀者收益​：

• 精通 List 類型，掌握其作為消息隊列、棧、最新列表的實現方法。
• 精通 Set 類型，掌握其去重特性和強大的集合運算（交集、並集、差集）。
• 精通 Sorted Set 類型，掌握其按分數排序的能力，輕鬆實現排行榜和範圍查詢。
• 能根據業務場景，在這三種結構間做出最合適的選擇。

​先修要求​：假設讀者已掌握 Redis 基礎連接和 String/Hash 操作（詳見系列前兩篇）。

​關鍵要點​：

1. List 是雙向鏈表，頭尾操作極快(O(1))，是實現簡單消息隊列的利器。
2. Set 保證元素唯一性，其集合運算能高效解決如“共同關注”等業務問題。
3. Sorted Set 兼具 Set 的唯一性和 List 的有序性，是排行榜功能的絕配。
4. 選擇正確的數據結構，往往比優化代碼更能提升系統性能和簡化開發。

背景與原理簡述

Redis 的魅力在於其為不同場景量身定製的數據結構。本篇主角是三種常用於實現複雜業務邏輯的類型：

• ​List（列表）​: 一個簡單的字符串列表，按插入順序排序。它是雙向鏈表實現的，這意味着在頭部和尾部添加或刪除元素的速度非常快（O(1)），但通過索引訪問中間元素較慢（O(n)）。
• ​Set（集合）​: Redis Set 是字符串的無序集合，通過哈希表實現。其最大特點是 元素唯一性 和 ​高效的集合運算​（求交集、並集、差集）。
• ​Sorted Set（有序集合，簡稱 ZSet）​: 這是 Redis 最富表現力的數據結構之一。它類似 Set，但每個元素都關聯一個 score（分數），元素按 score 進行排序。它通過**跳躍表（Skip List）** 和哈希表實現，保證了排序和高效訪問。

理解它們的底層實現，有助於我們預測其性能並做出業務選擇。

環境準備與快速上手

本篇所有示例基於以下連接客户端展開。

# filename: setup.py
import os
import redis
from redis import Redis

# 使用連接池創建客户端
pool = redis.ConnectionPool(
    host=os.getenv('REDIS_HOST', 'localhost'),
    port=int(os.getenv('REDIS_PORT', 6379)),
    password=os.getenv('REDIS_PASSWORD'),
    decode_responses=True,
    max_connections=10
)
r = Redis(connection_pool=pool)

# 簡單的連接測試
assert r.ping() is True
print("連接成功，開始操作 List, Set 和 Sorted Set！")

核心用法與代碼示例

List (列表) 操作

基本操作與應用場景：消息隊列、最新列表

# filename: list_operations.py
def list_basic_operations():
    """List 基本操作：隊列、棧、最新列表"""
    key = 'task_queue'

    # 1. 生產者：從左側/右側推送任務 (LPUSH/RPUSH)
    # LPUSH + RPOP 或 RPUSH + LPOP 可實現FIFO隊列
    r.lpush(key, 'task_video_1')  # 左推
    r.lpush(key, 'task_video_2')
    r.rpush(key, 'task_email_1')  # 右推
    print(f"All tasks after push: {r.lrange(key, 0, -1)}") # ['task_video_2', 'task_video_1', 'task_email_1']

    # 2. 消費者：從右側/左側彈出任務 (RPOP/LPOP)
    # 經典隊列模式：生產者LPUSH，消費者RPOP (或反之)
    task = r.rpop(key)  # 彈出最老的任務 'task_email_1'
    print(f"Processing task: {task}")

    # 3. 阻塞式彈出 (BRPOP/BLPOP) - 真正的隊列消費者
    # 應用場景：任務隊列工作者。如果沒有元素，會阻塞連接直到超時或有新元素。
    # result = r.brpop(key, timeout=30) # 阻塞30秒等待任務
    # if result:
    #     queue_name, task = result
    #     print(f"Got task from blocking pop: {task}")

    # 4. 獲取列表範圍 (LRANGE) 和 長度 (LLEN)
    tasks = r.lrange(key, 0, -1)  # 獲取所有元素，慎用！可能很大。
    length = r.llen(key)
    print(f"Remaining tasks ({length}): {tasks}")

    # 5. 修剪列表 (LTRIM) - 維護固定長度的最新列表
    # 應用場景：最新100條消息、最近登錄用户列表
    list_key = 'recent_logins'
    for user_id in range(1, 105):
        r.lpush(list_key, f'user_{user_id}')
    # 只保留最新的100條，修剪掉索引100之後的所有元素
    r.ltrim(list_key, 0, 99)
    print(f"Recent logins count: {r.llen(list_key)}") # 100

# 運行示例
list_basic_operations()

Set (集合) 操作

基本操作與應用場景：標籤系統、唯一性保證

# filename: set_operations.py
def set_basic_operations():
    """Set 基本操作：標籤、抽獎、唯一值存儲"""
    article_id = 42
    tags_key = f'article:{article_id}:tags'

    # 1. 添加和獲取成員 (SADD/SMEMBERS)
    # 應用場景：文章標籤、用户興趣
    r.sadd(tags_key, 'python', 'redis', 'database', 'tutorial')
    all_tags = r.smembers(tags_key)
    print(f"Article tags: {all_tags}")

    # 2. 隨機彈出/獲取成員 (SPOP/SRANDMEMBER)
    # 應用場景：抽獎、隨機推薦
    winner = r.spop(tags_key)  # 隨機移除並返回一個元素
    random_tag = r.srandmember(tags_key)  # 隨機獲取但不移除一個元素
    print(f"Popped winner: {winner}, Random tag: {random_tag}")

    # 3. 檢查成員是否存在 (SISMEMBER) 和 移除成員 (SREM)
    is_member = r.sismember(tags_key, 'python')
    print(f"Is 'python' a tag? {is_member}")
    r.srem(tags_key, 'database')  # 移除指定成員

    # 4. 集合運算的威力 - 社交系統的核心
    # 應用場景：共同好友(交集)、可能認識的人(差集)、全部興趣(並集)
    user_a_friends = {'user_b', 'user_c', 'user_d'}
    user_b_friends = {'user_a', 'user_c', 'user_e'}

    r.sadd('user:a:friends', *user_a_friends) # 解包集合
    r.sadd('user:b:friends', *user_b_friends)

    # 交集 (SINTER): 共同好友
    common_friends = r.sinter('user:a:friends', 'user:b:friends')
    print(f"Common friends of A and B: {common_friends}") # {'user_c'}

    # 並集 (SUNION): 所有好友
    all_friends = r.sunion('user:a:friends', 'user:b:friends')
    print(f"All friends: {all_friends}") # {'user_a', 'user_b', 'user_c', 'user_d', 'user_e'}

    # 差集 (SDIFF): A有但B沒有的好友
    a_unique_friends = r.sdiff('user:a:friends', 'user:b:friends')
    print(f"Friends only in A: {a_unique_friends}") # {'user_d'}

    # 將運算結果存儲到新key (SINTERSTORE/SUNIONSTORE/SDIFFSTORE)
    r.sinterstore('user:a_b:common_friends', 'user:a:friends', 'user:b:friends')

# 運行示例
set_basic_operations()

Sorted Set (有序集合) 操作

基本操作與應用場景：排行榜、優先隊列

# filename: zset_operations.py
def zset_basic_operations():
    """Sorted Set 基本操作：排行榜、按分範圍查詢"""
    leaderboard_key = 'game:leaderboard'

    # 1. 添加成員和分數 (ZADD)
    # 應用場景：遊戲積分榜、熱門文章排行
    r.zadd(leaderboard_key, {'player_a': 1000, 'player_b': 1500, 'player_c': 800, 'player_d': 1200})

    # 2. 按分數升序/降序獲取排名 (ZRANGE/ZREVRANGE)
    # 獲取前3名 (降序)
    top_3 = r.zrevrange(leaderboard_key, 0, 2, withscores=True) # withscores=True 返回分數
    print(f"Top 3 players: {top_3}") # [('player_b', 1500.0), ('player_d', 1200.0), ('player_a', 1000.0)]

    # 3. 獲取成員的排名和分數 (ZRANK/ZREVRANK/ZSCORE)
    player_b_rank = r.zrevrank(leaderboard_key, 'player_b')  # 降序排名，0是第一名
    player_b_score = r.zscore(leaderboard_key, 'player_b')
    print(f"Player B rank: {player_b_rank}, score: {player_b_score}")

    # 4. 按分數範圍查詢 (ZRANGEBYSCORE/ZREVRANGEBYSCORE)
    # 獲取分數在 1000 到 1300 之間的玩家
    mid_tier_players = r.zrangebyscore(leaderboard_key, 1000, 1300, withscores=True)
    print(f"Players with score between 1000-1300: {mid_tier_players}")

    # 5. 遞增成員分數 (ZINCRBY) - 原子操作，排行榜核心
    # 玩家A贏得一局遊戲，加50分
    new_score = r.zincrby(leaderboard_key, 50, 'player_a')
    print(f"Player A's new score: {new_score}")

    # 6. 獲取集合大小 (ZCARD) 和 分數區間內成員數 (ZCOUNT)
    total_players = r.zcard(leaderboard_key)
    players_above_1000 = r.zcount(leaderboard_key, 1000, '+inf')
    print(f"Total players: {total_players}, Players above 1000: {players_above_1000}")

# 運行示例
zset_basic_operations()

性能優化與容量規劃

數據結構選擇參考

大 Key 預警

• ​List​: 避免使用過長的 List（如數十萬元素）。LRANGE, LTRIM 等 O(n) 命令會變慢。
• ​Set / Sorted Set​: 避免巨大的集合。SMEMBERS, ZRANGE 等命令會阻塞服務。對於大數據集，考慮使用 SSCAN, ZSCAN 進行增量迭代（後續文章詳解）。
• ​通用規則​: 單個 Value 的大小不應超過 10KB，集合的元素數量應儘量控制在萬級以內。

內存優化

Sorted Set 在元素較少且分數較小時，也會使用一種叫做 ziplist 的緊湊編碼。可通過 zset-max-ziplist-entries 和 zset-max-ziplist-value 參數配置。

安全與可靠性

1. ​阻塞命令​: BLPOP, BRPOP 等阻塞命令會佔用一個連接。確保客户端庫（如 redis-py）的連接池配置了足夠的連接數（max_connections）來處理併發阻塞。
2. ​原子性​: ZINCRBY, SPOP 等命令都是原子操作，可以安全地在併發環境下使用。
3. 生產環境禁用 ​KEYS**: 再次強調，絕對不要使用 KEYS * 來查找模式匹配的 Key，尤其是在包含大量 Key 的生產環境中。使用 SCAN 命令代替。

常見問題與排錯

• SMEMBERS 或 ZRANGE 0 -1 導致服務變慢**: 這是遇到了大 Key 問題。立即使用 SSCAN 或 ZSCAN 進行替代，並考慮拆分大集合。
• BLPOP 一直阻塞**: 檢查超時參數設置，並確保有生產者往列表裏推送消息。
• ​Sorted Set 分數相同時的排序​: 當多個成員分數相同時，它們將按**字典序**排列。
• ZADD 的 XX 和 NX 選項**: NX 表示僅添加新成員，XX 表示僅更新已有成員。善用它們可以避免意外覆蓋或插入。

實戰案例/最佳實踐

案例一：可靠的簡單消息隊列

# filename: simple_queue.py
import threading
import time

class SimpleTaskQueue:
    def __init__(self, redis_client, queue_name):
        self.r = redis_client
        self.queue_name = queue_name

    def produce_task(self, task_data):
        """生產者：推送任務"""
        # 使用 LPUSH 將任務推入隊列左側
        self.r.lpush(self.queue_name, task_data)
        print(f"Produced task: {task_data}")

    def consume_task(self, worker_id):
        """消費者：阻塞地獲取並處理任務"""
        print(f"Worker {worker_id} started...")
        while True:
            # 使用 BRPOP 從隊列右側阻塞地獲取任務，超時時間5秒
            # BRPOP 返回一個元組 (list_name, element)
            result = self.r.brpop(self.queue_name, timeout=5)
            if result is None:
                # 超時，沒有任務，可以做一些其他工作或繼續循環
                print(f"Worker {worker_id}: No task, waiting...")
                continue

            queue_name, task_data = result
            print(f"Worker {worker_id} processing task: {task_data}")
            # 模擬任務處理時間
            time.sleep(1)
            print(f"Worker {worker_id} finished task: {task_data}")

# 使用示例
def demo_queue():
    queue = SimpleTaskQueue(r, 'my_task_queue')

    # 啓動一個消費者線程
    consumer_thread = threading.Thread(target=queue.consume_task, args=('worker_1',), daemon=True)
    consumer_thread.start()

    # 主線程生產任務
    for i in range(5):
        queue.produce_task(f'task_data_{i}')
        time.sleep(0.5)

    time.sleep(6) # 讓消費者有時間處理

# demo_queue()

案例二：實時遊戲排行榜

# filename: game_leaderboard.py
class GameLeaderboard:
    def __init__(self, redis_client, leaderboard_key):
        self.r = redis_client
        self.leaderboard_key = leaderboard_key

    def add_player(self, player_id, initial_score=0):
        """添加玩家到排行榜"""
        self.r.zadd(self.leaderboard_key, {player_id: initial_score})

    def update_score(self, player_id, delta):
        """更新玩家分數（增加或減少）"""
        new_score = self.r.zincrby(self.leaderboard_key, delta, player_id)
        return new_score

    def get_top_n(self, n=10):
        """獲取前N名玩家"""
        return self.r.zrevrange(self.leaderboard_key, 0, n-1, withscores=True)

    def get_player_rank_and_score(self, player_id):
        """獲取玩家的排名和分數"""
        rank = self.r.zrevrank(self.leaderboard_key, player_id)
        if rank is None:
            return None, None
        score = self.r.zscore(self.leaderboard_key, player_id)
        return rank + 1, score # 排名從0開始，轉為從1開始更直觀

    def get_players_around_me(self, player_id, range_size=2):
        """獲取玩家附近的競爭對手（前後各range_size名）"""
        rank = self.r.zrevrank(self.leaderboard_key, player_id)
        if rank is None:
            return []

        start = max(0, rank - range_size)
        end = rank + range_size
        return self.r.zrevrange(self.leaderboard_key, start, end, withscores=True)

# 使用示例
def demo_leaderboard():
    lb = GameLeaderboard(r, 'my_game_leaderboard')
    players = ['player_1', 'player_2', 'player_3', 'player_4']

    # 初始化
    for p in players:
        lb.add_player(p)

    # 模擬遊戲更新
    lb.update_score('player_1', 100)
    lb.update_score('player_2', 150)
    lb.update_score('player_3', 75)
    lb.update_score('player_4', 200)
    lb.update_score('player_1', 50) # player_1 又得了50分

    # 查詢
    top_2 = lb.get_top_n(2)
    print(f"Top 2: {top_2}")

    rank, score = lb.get_player_rank_and_score('player_1')
    print(f"Player_1 rank: {rank}, score: {score}")

    around = lb.get_players_around_me('player_1', 1)
    print(f"Around player_1: {around}")

demo_leaderboard()

小結

List、Set 和 Sorted Set 將 Redis 從簡單的鍵值存儲提升為了一個強大的數據結構和計算平台。List 提供了順序和阻塞操作，Set 提供了唯一性和集合運算，Sorted Set 提供了排序和範圍查詢。