Redis 数据类型完全指南：String/Hash/List/Set/ZSet 深入理解

小飞兽 Redis 4 次阅读 2026-07-29

前言

Redis 提供了丰富的数据类型，每种类型都有其最佳使用场景。理解这些数据类型的底层实现（SDS、ziplist、quicklist、intset、dict、skiplist），才能真正用好 Redis。本文系统讲解五种核心类型的底层原理与实战技巧。

String 类型：最常用的数据结构

Redis 的 String 类型并非简单的 C 字符串，而是使用了 SDS（Simple Dynamic String）动态字符串。SDS 解决了 C 字符串的诸多问题：长度计算 O(n)、缓冲区溢出、二进制不安全。

SDS 的特点：

<code>len</code> 字段记录字符串长度，strlen 为 O(1)

自动扩容，容量不足时翻倍扩展（小于 1MB 时）或增加 1MB（大于 1MB 时）

支持二进制数据（可以存储 JPEG、JSON 等任意字节序列）

惰性释放：字符串缩短时不立即回收内存

import redis

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

# 基本操作
r.set('user:1001:name', '张三')
r.set('user:1001:age', 28)
r.get('user:1001:name')  -- '张三'

# SET 的高级选项
r.set('token:abc123', 'user_id_1001', ex=3600, nx=True)  -- nx=True: 不存在才设置（原子）
r.set('rate:limit:api', 100, ex=60, xx=True)  -- xx=True: 存在才更新

# 批量操作
r.mset({
    'product:1001:price': '299.00',
    'product:1001:stock': '500',
    'product:1001:name': '无线蓝牙耳机'
})
r.mget(['product:1001:price', 'product:1001:stock'])
-- ['299.00', '500']

# 计数操作（原子递增/递减）
r.set('page:views:20240101', 1000)
r.incr('page:views:20240101')       -- 1001
r.incrby('page:views:20240101', 50)  -- 1051
r.decr('page:views:20240101')       -- 1050
r.incrbyfloat('price:discount', 0.5)

# 字符串切片（Redis 6.2+）
r.setrange('log', 0, 'prefix:')
r.getrange('log', 0, 10)

String 类型的应用场景：

缓存序列化的 JSON 对象（但大对象推荐用 Hash）

计数器（访问次数、点赞数）

分布式锁（SET + NX + EX 组合）

Session 存储

限流（令牌桶/滑动窗口）

Hash 类型：字段级操作

Hash 是一个 field-value 映射表，适合存储对象。相比 String + 序列化 JSON 的方案，Hash 可以对单个字段进行独立操作，无需反序列化整个对象。

# 存储用户对象（Hash vs String 的对比）
# String 方案：整个对象序列化存储
r.set('user:1001', json.dumps({'name': '张三', 'age': 28}))
data = json.loads(r.get('user:1001'))
data['age'] = 29  -- 需要：反序列化 -> 修改 -> 重新序列化
r.set('user:1001', json.dumps(data))

# Hash 方案：字段级操作，无需序列化
r.hset('user:1001', mapping={'name': '张三', 'age': 28, 'city': '北京'})
r.hget('user:1001', 'age')       -- '28'
r.hgetall('user:1001')            -- {'name': '张三', 'age': '28', 'city': '北京'}
r.hincrby('user:1001', 'age', 1)  -- 原子递增：age = 29
r.hset('user:1001', 'email', 'zhangsan@example.com')
r.hdel('user:1001', 'city')
r.hexists('user:1001', 'name')   -- True

# 批量操作
r.hmset('product:2001', {'name': '无线键盘', 'price': '199.00', 'stock': '200'})
r.hmget('product:2001', ['name', 'price'])
r.hlen('product:2001')           -- 3
r.hkeys('product:2001')
r.hvals('product:2001')

# 游标扫描（大数据量时避免阻塞）
for key, fields in r.hscan_iter('large:hash:key', count=1000):
    process(fields)

Hash 的底层实现：

数据量小（默认小于 512 个字段，每个字段值小于 64 字节）时使用 ziplist（压缩列表），内存紧凑

超过阈值后转换为 dict（哈希表），支持 O(1) 随机访问

配置项：<code>hash-max-ziplist-entries</code> 和 <code>hash-max-ziplist-value</code>

List 类型：有序队列

Redis List 是双向链表实现的有序序列，支持两端操作（LPUSH/RPOP 等），常用于消息队列、最新列表等场景。

# 基础操作
r.lpush('queue:tasks', 'task:001', 'task:002', 'task:003')
r.rpush('queue:tasks', 'task:004')
r.llen('queue:tasks')           -- 4
r.lrange('queue:tasks', 0, -1)  -- ['task:003', 'task:002', 'task:001', 'task:004']
r.lpop('queue:tasks')           -- 'task:003'
r.rpop('queue:tasks')           -- 'task:004'

# 阻塞操作（实现消息队列）
result = r.blpop('queue:tasks', timeout=30)  -- 返回 (key, value) 或 None

# 列表切片
r.rpush('article:ids', 'a1', 'a2', 'a3', 'a4', 'a5')
r.lrange('article:ids', 0, 2)
r.ltrim('article:ids', 0, 9)    -- 裁剪保留 [0, 9]

# 实现最新列表（如微博最新 100 条评论）
def add_comment(article_id, comment_id):
    pipe = r.pipeline()
    pipe.lpush(f'comments:{article_id}', comment_id)
    pipe.ltrim(f'comments:{article_id}', 0, 99)
    pipe.execute()

# 列表间移动
r.rpoplpush('queue:tasks:processing', 'queue:tasks:completed')
r.brpoplpush('queue:tasks', 'queue:tasks:backup', timeout=10)

List 的应用场景：

消息队列（轻量级，生产环境推荐用 Stream）

最新列表（最新 N 条评论/帖子）

时间线（用户动态列表）

栈和队列数据结构

Set 类型：无序去重集合

Set 是无序、去重、元素唯一的集合。底层使用 intset（整数集合，小数据量时）或 dict（哈希表）实现，支持交集、并集、差集等集合运算。

# 基础操作
r.sadd('tags:article:1001', 'Redis', 'MySQL', 'Python', '数据库')
r.scard('tags:article:1001')      -- 4
r.smembers('tags:article:1001')
r.sismember('tags:article:1001', 'Redis')  -- True
r.sismember('tags:article:1001', 'MongoDB')  -- False

# 随机操作
r.srandmember('tags:article:1001')
r.srandmember('tags:article:1001', 2)
r.spop('tags:article:1001')

# 集合运算
r.sadd('users:viewed:20240101', 'u1', 'u2', 'u3', 'u4')
r.sadd('users:viewed:20240102', 'u3', 'u4', 'u5', 'u6')

-- 差集：昨天看了今天没看的（流失用户）
r.sdiff('users:viewed:20240101', 'users:viewed:20240102')

-- 交集：两天都看了的（活跃用户）
r.sinter('users:viewed:20240101', 'users:viewed:20240102')

-- 并集：两天合计看过的人数（去重）
r.sunion('users:viewed:20240101', 'users:viewed:20240102')

-- 将集合运算结果存储到新 key
r.sinterstore('users:active', 'users:viewed:20240101', 'users:viewed:20240102')

# 遍历大集合
for member in r.sscan_iter('large:set:key', count=1000):
    process(member)

Set 的应用场景：

标签系统（文章标签、用户画像标签）

点赞、收藏、去重（用户 ID 集合）

关注列表（但无法按时间排序，ZSet 更适合）

黑名单/白名单

ZSet 类型：有序去重集合

ZSet（Sorted Set）是 Redis 最复杂的数据类型，每个元素有一个 score（浮点数），按 score 排序。底层使用 dict（哈希表）+ skiplist（跳表）双重索引，查找、插入、删除都是 O(log N)。

为什么用跳表而不是 B+ 树？
跳表实现简单，易于理解和实现。插入节点时只需局部修改指针，无需像 B+ 树那样可能触发节点分裂。跳表的平均复杂度为 O(log N)，实际表现非常稳定。

# 基础操作
r.zadd('leaderboard:posts', {'post:1001': 1500, 'post:1002': 2300, 'post:1003': 800})
r.zcard('leaderboard:posts')         -- 3
r.zscore('leaderboard:posts', 'post:1001')  -- 1500.0
r.zrank('leaderboard:posts', 'post:1001')   -- 2（0-indexed）
r.zrevrank('leaderboard:posts', 'post:1001')  -- 0

# 按排名范围查询
r.zrange('leaderboard:posts', 0, 9, withscores=True)
r.zrevrange('leaderboard:posts', 0, 9, withscores=True)

# 按分数范围查询
r.zrangebyscore('leaderboard:posts', 1000, 2000, withscores=True)
r.zcount('leaderboard:posts', 1000, 2000)

# 更新分数（原子操作）
r.zincrby('leaderboard:posts', 100, 'post:1001')

# 删除元素
r.zrem('leaderboard:posts', 'post:1003')

# 按排名删除（分页场景）
r.zremrangebyrank('leaderboard:posts', 0, -11)

# 按分数删除
r.zremrangebyscore('leaderboard:posts', 0, 500)

# 应用：微博热搜排行榜
def on_post_like(post_id):
    r.zincrby('hot:posts:daily', 1, str(post_id))
    r.expire('hot:posts:daily', 86400)

def get_hot_posts(limit=10):
    return r.zrevrange('hot:posts:daily', 0, limit - 1, withscores=True)

# 应用：延时队列
import time
def add_delay_task(task_id, delay_seconds):
    r.zadd('delay:queue', {task_id: time.time() + delay_seconds})

def poll_delay_task():
    now = time.time()
    tasks = r.zrangebyscore('delay:queue', 0, now)
    if tasks:
        r.zrem('delay:queue', *tasks)
    return tasks

常见问题

Q1：String 类型的值最大能存多大？
Redis String 类型最大支持 512MB。存储大文件（如图片 Base64）应谨慎，建议上传到对象存储，将 URL 存入 Redis。

Q2：Hash 有多少字段会触发 ziplist -> dict 的转换？
默认配置下，当字段数超过 hash-max-ziplist-entries（默认 512）或任一字段值超过 hash-max-ziplist-value（默认 64 字节）时，会转换为 dict。

Q3：ZSet 的 score 相同时如何排序？
当 score 相同时，Redis 按字典序（lexicographical order）排列元素。这使得 ZSet 可以实现字典序范围查询（ZRANGEBYLEX），常用于实现拼音搜索、电话号码簿等场景。