Redis ZSet与SkipList跳表

Redis 的五种数据结构各有特色，用对了才能发挥它的优势。很多人只用到了 String 和 Hash，却不知道 List、Set、ZSet 在特定场景下更合适。本文从应用场景出发，讲什么时候用什么类型。

一、ZSet的编码方式

#

1.1 两种编码

编码	结构	适用条件
ziplist	压缩列表	元素少且值小
skiplist	跳表+哈希表	元素多或值大

#

1.2 编码转换条件

zset-max-ziplist-entries 128    # ziplist最多128个entry
zset-max-ziplist-value 64       # 每个member最大64字节

转换规则：

• 元素数量<=128且每个member<=64字节：ziplist
• 任一条件不满足：skiplist

二、SkipList原理

#

2.1 为什么用SkipList

SkipList是一种概率平衡的数据结构：

数据结构	查找	插入	删除	范围查询	实现复杂度
数组	O(n)	O(n)	O(n)	O(1)开始	低
链表	O(n)	O(1)	O(1)	O(n)	低
平衡树	O(log n)	O(log n)	O(log n)	复杂	高
SkipList	O(log n)	O(log n)	O(log n)	O(log n)开始	低

SkipList的优势：

• 实现简单（比红黑树简单得多）
• 范围查询友好（比树结构简单）
• 插入/删除不需要旋转平衡
• 性能接近平衡树

#

2.2 SkipList结构

Level 3:  head ──────────────────────────>  NULL

Level 2:  head ──────────────>  30 ──────>  NULL

Level 1:  head ──────>  20 ──>  30 ──────>  NULL

Level 0:  head ──> 10 ──> 20 ──> 30 ──> 40 ──> 50 ──> NULL
                (1, "a")  (2, "b")  (3, "c")  (4, "d")  (5, "e")
                
每个节点有多个level的forward指针

#

2.3 Redis SkipList节点

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;

#

2.4 随机层数生成

int zslRandomLevel(void) {
    int level = 1;
    // 每次有25%概率升级一层
    while ((random() & 0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;
    // 最大32层
    return (level < ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

// ZSKIPLIST_P = 0.25
// 层数概率分布：
// level=1: 75%
// level=2: 18.75%
// level=3: 4.6875%
// ...

#

2.5 查找操作

查找score=35的节点：

Level 3: head ──────────────────────────> NULL
              │ 比较30<35，继续
              ▼
Level 2: head ──────────────> 30 ───────> NULL
                              │ 比较30<35，继续
                              ▼
Level 1: head ──────> 20 ──> 30 ────────> NULL
                                    │ 下一个NULL，下降
                                    ▼
Level 0: ... ──> 30 ──> 40 ──> 50 ──> NULL
              │         │
              30<35    40>35
              下降     找到位置（在30和40之间）

查找复杂度：O(log n)

#

2.6 插入操作

zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;
    
    // 1. 查找每一层的前驱节点
    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        rank[i] = (i == zsl->level-1) ? 0 : rank[i+1];
        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;
    }
    
    // 2. 随机生成层数
    level = zslRandomLevel();
    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;
        }
        zsl->level = level;
    }
    
    // 3. 创建新节点
    x = zslCreateNode(level, score, ele);
    
    // 4. 插入到各层链表
    for (i = 0; i < level; i++) {
        x->level[i].forward = update[i]->level[i].forward;
        update[i]->level[i].forward = x;
        
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }
    
    // 5. 更新高层span
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++;
    }
    
    // 6. 设置后退指针
    x->backward = (update[0] == zsl->header) ? NULL : update[0];
    if (x->level[0].forward)
        x->level[0].forward->backward = x;
    else
        zsl->tail = x;
    
    zsl->length++;
    return x;
}

#

2.7 span（跨度）的作用

span记录当前节点到forward节点之间跳过了多少个节点：

Level 2:  head ──────────────────>  40
           span=3                    
           （跳过了3个level0节点）

Level 1:  head ──────────>  30 ───>  40
           span=2              span=1
           
Level 0:  head ──> 10 ──> 20 ──> 30 ──> 40
           span=1  span=1  span=1  span=1

获取排名：
ZRANK key 40 = 1+1+1 = 3（从0开始）

三、哈希表的作用

#

3.1 为什么需要哈希表

typedef struct zset {
    dict *dict;           // member -> score 的哈希表
    zskiplist *zsl;       // 跳表
} zset;

SkipList按score排序，但查找特定member需要O(n)。哈希表提供O(1)的member查找。

#

3.2 两种结构的配合

操作          使用结构          原因
─────────────────────────────────────────────
ZADD          两者都用          插入到跳表，更新哈希表
ZSCORE        哈希表            O(1)获取score
ZRANK         跳表              按score排序
ZRANGE        跳表              范围查询
ZREM          两者都用          从两个结构中删除
ZINCRBY       两者都用          修改score
ZCARD         跳表length        O(1)

四、ZSet命令实现

#

4.1 ZADD

void zaddCommand(client *c) {
    robj *key = c->argv[1];
    robj *zobj = lookupKeyWrite(c->db, key);
    
    // 创建或获取ZSet
    if (zobj == NULL) {
        zobj = createZsetObject();  // 创建skiplist+hashtable
        dbAdd(c->db, key, zobj);
    }
    
    // 添加元素
    int added = 0;
    for (int j = 2; j < c->argc; j += 2) {
        double score = strtod(c->argv[j]->ptr, NULL);
        sds ele = c->argv[j+1]->ptr;
        
        int retval = zsetAdd(zobj, score, ele, &added);
    }
    
    addReplyLongLong(c, added);
}

#

4.2 ZRANK

void zrankCommand(client *c) {
    robj *zobj = lookupKeyReadOrReply(c, c->argv[1], shared.null[c->resp]);
    if (zobj == NULL || checkType(c, zobj, OBJ_ZSET)) return;
    
    // 先查哈希表获取score
    double score;
    if (zsetScore(zobj, c->argv[2]->ptr, &score) == C_ERR) {
        addReplyNull(c);
        return;
    }
    
    // 在跳表中查找排名
    long rank = zsetRank(zobj, c->argv[2]->ptr, score, 1);
    addReplyLongLong(c, rank);
}

#

4.3 ZRANGE

void zrangeCommand(client *c) {
    robj *zobj = lookupKeyReadOrReply(c, c->argv[1], shared.emptyarray);
    if (zobj == NULL || checkType(c, zobj, OBJ_ZSET)) return;
    
    long start, end;
    getLongFromObjectOrReply(c, c->argv[2], &start, NULL);
    getLongFromObjectOrReply(c, c->argv[3], &end, NULL);
    
    // 处理负数索引
    long llen = zsetLength(zobj);
    if (start < 0) start = llen + start;
    if (end < 0) end = llen + end;
    
    // 遍历跳表指定范围
    zset *zs = zobj->ptr;
    zskiplistNode *ln = zs->zsl->header->level[0].forward;
    
    // 跳到start位置
    while (start-- && ln) ln = ln->level[0].forward;
    
    // 输出范围内的元素
    addReplyArrayLen(c, rangelen);
    while (rangelen-- && ln) {
        addReplyBulkCBuffer(c, ln->ele, sdslen(ln->ele));
        if (withscores) addReplyDouble(c, ln->score);
        ln = ln->level[0].forward;
    }
}

五、应用场景

#

5.1 排行榜

@Service
public class LeaderboardService {
    
    private static final String LEADERBOARD = "leaderboard:weekly";
    
    // 增加积分
    public void addScore(Long userId, double score) {
        redis.opsForZSet().incrementScore(LEADERBOARD, String.valueOf(userId), score);
    }
    
    // 获取前10名
    public Set<ZSetOperations.TypedTuple<String>> getTop10() {
        return redis.opsForZSet().reverseRangeWithScores(LEADERBOARD, 0, 9);
    }
    
    // 获取用户排名（从0开始）
    public Long getRank(Long userId) {
        Long rank = redis.opsForZSet().reverseRank(LEADERBOARD, String.valueOf(userId));
        return rank != null ? rank + 1 : null;  // 转为1-based
    }
    
    // 获取用户积分
    public Double getScore(Long userId) {
        return redis.opsForZSet().score(LEADERBOARD, String.valueOf(userId));
    }
    
    // 获取用户附近排名
    public Set<ZSetOperations.TypedTuple<String>> getNearbyRank(Long userId) {
        Long rank = redis.opsForZSet().reverseRank(LEADERBOARD, String.valueOf(userId));
        if (rank == null) return Collections.emptySet();
        
        long start = Math.max(0, rank - 5);
        long end = rank + 5;
        return redis.opsForZSet().reverseRangeWithScores(LEADERBOARD, start, end);
    }
    
    // 重置排行榜
    public void resetLeaderboard() {
        redis.delete(LEADERBOARD);
    }
}

#

5.2 延时队列

@Component
public class DelayQueue {
    
    @Autowired
    private StringRedisTemplate redis;
    
    // 添加延时任务
    public void addDelayTask(String taskId, long delayMs) {
        long executeTime = System.currentTimeMillis() + delayMs;
        redis.opsForZSet().add("delay:queue", taskId, executeTime);
    }
    
    // 消费到期任务
    @Scheduled(fixedRate = 1000)
    public void consumeDelayTasks() {
        long now = System.currentTimeMillis();
        
        // 获取到期的任务（score <= now）
        Set<String> tasks = redis.opsForZSet()
            .rangeByScore("delay:queue", 0, now, 0, 100);
        
        for (String taskId : tasks) {
            // 删除并处理（保证幂等）
            Long removed = redis.opsForZSet().remove("delay:queue", taskId);
            if (removed != null && removed > 0) {
                processTask(taskId);
            }
        }
    }
    
    private void processTask(String taskId) {
        // 处理任务
        System.out.println("Processing task: " + taskId);
    }
}

#

5.3 滑动窗口限流

@Component
public class SlidingWindowRateLimiter {
    
    public boolean isAllowed(String userId, int maxRequests, long windowMs) {
        String key = "rate_limit:" + userId;
        long now = System.currentTimeMillis();
        long windowStart = now - windowMs;
        
        // 移除窗口外的请求记录
        redis.opsForZSet().removeRangeByScore(key, 0, windowStart);
        
        // 获取当前窗口内的请求数
        Long count = redis.opsForZSet().zCard(key);
        if (count != null && count >= maxRequests) {
            return false;
        }
        
        // 记录本次请求（使用雪花ID或UUID保证member唯一）
        redis.opsForZSet().add(key, UUID.randomUUID().toString(), now);
        redis.expire(key, windowMs / 1000 + 1, TimeUnit.SECONDS);
        
        return true;
    }
}

#

5.4 时间线/Feed流

@Service
public class FeedService {
    
    // 发布内容（score为发布时间戳）
    public void postContent(Long userId, String contentId) {
        long timestamp = System.currentTimeMillis();
        redis.opsForZSet().add("feed:user:" + userId, contentId, timestamp);
        redis.opsForZSet().removeRange("feed:user:" + userId, 0, -1001);  // 保留1000条
    }
    
    // 获取最新内容（分页）
    public Set<String> getFeed(Long userId, int page, int size) {
        long start = (long) page * size;
        long end = start + size - 1;
        return redis.opsForZSet().reverseRange("feed:user:" + userId, start, end);
    }
    
    // 获取某时间范围内的内容
    public Set<String> getFeedByTimeRange(Long userId, long startTime, long endTime) {
        return redis.opsForZSet().reverseRangeByScore(
            "feed:user:" + userId, startTime, endTime);
    }
}

六、性能优化

#

6.1 ZSet大小控制

// 只保留最新的N条
redis.opsForZSet().removeRange("zset", 0, -1001);

// 定期清理过期数据
redis.opsForZSet().removeRangeByScore("zset", 0, System.currentTimeMillis() - 86400000);

#

6.2 批量操作

// 批量添加
Set<ZSetOperations.TypedTuple<String>> tuples = new HashSet<>();
for (UserScore us : userScores) {
    tuples.add(new DefaultTypedTuple<>(us.getUserId().toString(), us.getScore()));
}
redis.opsForZSet().add("leaderboard", tuples);

#

6.3 避免大ZSet

// 大ZSet分片
public void addToShardZSet(String key, String member, double score) {
    int shard = member.hashCode() % 10;
    redis.opsForZSet().add(key + ":" + shard, member, score);
}

七、常见问题

#

7.1 ZSet最大元素数

理论限制：约40亿（2^32 - 1）
实际限制：内存大小

建议：单个ZSet不超过10000个元素

#

7.2 score相同时的排序

# score相同时，按member的字典序排序
ZADD myzset 1 "b"
ZADD myzset 1 "a"
ZRANGE myzset 0 -1
# 返回: a, b（字典序）

#

7.3 分数精度

# Redis使用double存储score，有精度限制
ZADD myzset 9007199254740992 "a"
ZADD myzset 9007199254740993 "b"
# 这两个分数可能被视为相同

# 解决：使用整数score（如时间戳*1000 + 序列号）

八、总结

特性	SkipList
查找	O(log n)
插入	O(log n)
删除	O(log n)
范围查询	O(log n + m)
实现复杂度	低

ZSet核心设计：

• SkipList：按score排序，支持范围查询和排名
• Hashtable：按member查找，O(1)获取score
• 双向链表：支持倒序遍历

ZSet使用建议：

1. score设计：使用整数避免精度问题
2. 控制大小：定期清理，避免无限增长
3. 批量操作：减少网络往返
4. 范围查询优先：ZRANGE比ZRANK更高效
5. 考虑Stream：需要消费者组时用Stream替代延时队列

核心要点

1. String：简单的键值对，适合缓存、计数器

2. Hash：存储对象属性，适合用户信息、配置

3. List：有序列表，适合消息队列、最新列表

4. Set：无序去重，适合共同好友、抽奖

5. ZSet：有序集合，适合排行榜、积分系统

总结

选择合适的数据结构是使用 Redis 的关键。在实际项目中，根据业务需求选择合适的类型，可以提升性能和开发效率。