Redis使用简明教程

Redis 是一款高性能的键值对存储数据库，支持丰富的数据结构和高级特性。下面结合 Python SDK（redis-py 库），分别介绍其通用用法和高级用法，并附具体代码示例。

一、通用用法（基础数据结构操作）

Redis 提供了 5 种核心数据结构，适用于大多数日常场景（如缓存、计数器、简单队列等）。

1. 字符串（String）

最基础的键值对类型，适用于存储文本、数字等，支持原子操作。

import redis# 连接 Redis（默认 host=localhost, port=6379, db=0）
r = redis.Redis(decode_responses=True)  # decode_responses=True 自动将 bytes 转为字符串# 设置键值（ex 为过期时间，单位秒）
r.set("name", "Alice", ex=3600)  # 1小时后过期
r.set("age", 25)# 获取值
print(r.get("name"))  # 输出: Alice
print(r.get("age"))   # 输出: 25# 原子递增/递减（适用于计数器）
r.incr("age")  # age 变为 26
r.decr("age")  # age 变回 25

2. 哈希（Hash）

适用于存储对象（如用户信息、商品属性），键值对的嵌套结构。

# 存储用户信息（hset 支持映射或键值对）
r.hset("user:100", mapping={"name": "Bob","age": 30,"city": "Shanghai"
})# 获取单个字段
print(r.hget("user:100", "name"))  # 输出: Bob# 获取所有字段和值
print(r.hgetall("user:100"))  # 输出: {'name': 'Bob', 'age': '30', 'city': 'Shanghai'}# 检查字段是否存在
print(r.hexists("user:100", "email"))  # 输出: False

3. 列表（List）

有序可重复的元素集合，适用于实现队列、栈、最新消息列表等。

# 左侧插入元素（lpush 左进，rpush 右进）
r.lpush("tasks", "task1", "task2")  # 列表变为: [task2, task1]# 右侧插入元素
r.rpush("tasks", "task3")  # 列表变为: [task2, task1, task3]# 获取指定范围元素（0 为开头，-1 为结尾）
print(r.lrange("tasks", 0, -1))  # 输出: ['task2', 'task1', 'task3']# 弹出元素（lpop 左出，rpop 右出）
print(r.lpop("tasks"))  # 输出: task2（列表剩余: [task1, task3]）

4. 集合（Set）

无序不可重复的元素集合，适用于去重、交集/并集计算（如共同好友）。

# 添加元素
r.sadd("tags", "python", "redis", "database")# 获取所有元素
print(r.smembers("tags"))  # 输出: {'python', 'redis', 'database'}（无序）# 检查元素是否存在
print(r.sismember("tags", "java"))  # 输出: False# 交集计算（如用户1和用户2的共同关注）
r.sadd("user:1:follows", "A", "B", "C")
r.sadd("user:2:follows", "B", "C", "D")
print(r.sinter("user:1:follows", "user:2:follows"))  # 输出: {'B', 'C'}

5. 有序集合（Sorted Set）

有序且不可重复的元素集合，每个元素关联一个分数（score），适用于排行榜、优先级队列等。

# 添加元素（值 + 分数）
r.zadd("leaderboard", {"user1": 90,"user2": 100,"user3": 80
})# 按分数升序获取元素（0 为最低分，-1 为最高分）
print(r.zrange("leaderboard", 0, -1, withscores=True))  
# 输出: [('user3', 80.0), ('user1', 90.0), ('user2', 100.0)]# 按分数降序获取 Top 2
print(r.zrevrange("leaderboard", 0, 1, withscores=True))  
# 输出: [('user2', 100.0), ('user1', 90.0)]# 统计分数在 80-100 之间的元素数量
print(r.zcount("leaderboard", 80, 100))  # 输出: 3

二、高级用法（进阶特性）

Redis 提供了多种高级特性，适用于性能优化、复杂业务场景（如分布式锁、消息队列、实时统计等）。

1. 事务（Transaction）

通过 MULTI/EXEC 保证一系列操作的原子性（要么全执行，要么全不执行）。

# 示例：转账操作（A 减 100，B 加 100，需原子执行）
with r.pipeline() as pipe:  # pipeline 用于事务和批量操作try:# 监听键，若期间被修改则事务中断pipe.watch("balance:A", "balance:B")# 开始事务pipe.multi()pipe.decrby("balance:A", 100)pipe.incrby("balance:B", 100)# 执行事务pipe.execute()print("转账成功")except redis.WatchError:# 键被修改，事务失败print("转账失败，数据已变更")

2. 管道（Pipeline）

批量发送命令，减少网络往返次数，提升性能（事务依赖管道实现）。

# 批量执行多个命令
with r.pipeline() as pipe:# 链式调用多个命令pipe.set("a", 1).set("b", 2).get("a").get("b")# 一次性执行，返回结果列表results = pipe.execute()print(results)  # 输出: [True, True, '1', '2']

3. 发布订阅（Pub/Sub）

实现消息通信模式（生产者发布消息，消费者订阅频道接收消息）。

# 订阅者（持续监听消息）
def subscriber():pubsub = r.pubsub()pubsub.subscribe("news")  # 订阅 "news" 频道for message in pubsub.listen():# 过滤非消息类型（如订阅确认）if message["type"] == "message":print(f"收到消息: {message['data']}")# 发布者（发送消息）
def publisher():r.publish("news", "Redis 发布订阅示例")  # 向 "news" 频道发布消息# 实际使用时需多线程/进程运行（订阅者阻塞）
import threading
threading.Thread(target=subscriber, daemon=True).start()
publisher()  # 输出: 收到消息: Redis 发布订阅示例

4. Lua 脚本

通过 Lua 脚本执行复杂逻辑，保证原子性（脚本内的操作不会被其他命令中断）。

# 示例：判断键是否存在，存在则返回值，否则设置默认值
script = """
local key = KEYS[1]
local default = ARGV[1]
if redis.call('exists', key) == 1 thenreturn redis.call('get', key)
elseredis.call('set', key, default)return default
end
"""# 执行脚本（1 个键，1 个参数）
result = r.eval(script, 1, "config", "default_value")
print(result)  # 首次执行输出: default_value（键不存在时）

5. Stream（消息队列）

比 Pub/Sub 更强大的消息队列，支持消息持久化、消费组（避免重复消费）。

# 生产者：添加消息到 Stream（id="*" 表示自动生成ID）
r.xadd("order_stream", {"order_id": "1001", "user": "Alice"}, id="*")
r.xadd("order_stream", {"order_id": "1002", "user": "Bob"}, id="*")# 消费者组：创建消费组（首次需初始化）
r.xgroup_create("order_stream", "group1", id="$")  # "$" 表示从最新消息开始消费# 消费者：从组内获取消息（count=2 表示最多2条）
messages = r.xreadgroup(groupname="group1",consumername="consumer1",streams={"order_stream": ">"},  # ">" 表示未被消费的消息count=2
)
print(messages)  # 输出消息列表# 确认消息已处理（避免重复消费）
for msg_id in [msg[0] for msg in messages[0][1]]:r.xack("order_stream", "group1", msg_id)

6. 地理空间（GEO）

存储地理位置信息，支持距离计算、附近点查询（如“附近的门店”）。

# 添加地理位置（经度, 纬度, 名称）
r.geoadd("cities", (116.4074, 39.9042, "beijing"),  # 北京(121.4737, 31.2304, "shanghai")  # 上海
)# 计算两地距离（单位 km）
distance = r.geodist("cities", "beijing", "shanghai", unit="km")
print(f"北京到上海距离: {distance:.2f} km")  # 输出约 1318.39 km# 查询指定坐标附近的点（半径 2000 km，最多 10 个）
nearby = r.georadius("cities", 116.4074, 39.9042, 2000, "km", count=10, withdist=True)
print(nearby)  # 输出: [('beijing', 0.0), ('shanghai', 1318.3933)]

7. 位图（Bitmap）

用位存储数据，适用于高效统计（如用户签到、活跃用户计数）。

# 记录用户签到（用户ID=10086 在第 5 天签到）
r.setbit("user:sign:10086", 5, 1)  # 第 5 位设为 1（从 0 开始）# 检查用户是否在第 5 天签到
print(r.getbit("user:sign:10086", 5))  # 输出: 1# 统计用户总签到天数（位为 1 的数量）
print(r.bitcount("user:sign:10086"))  # 输出: 1

8. HyperLogLog

用于超大数据量的基数统计（如独立访客 UV 计数），占用内存极小。

# 添加访问用户
r.pfadd("uv:20231001", "user1", "user2", "user3")
r.pfadd("uv:20231001", "user3", "user4", "user5")  # 去重统计# 估算独立用户数（误差约 0.8%）
print(r.pfcount("uv:20231001"))  # 输出: 5# 合并多天的 UV 统计
r.pfmerge("uv:202310", "uv:20231001", "uv:20231002")

总结

• 通用用法：基于 5 种核心数据结构，满足缓存、计数、简单队列等基础需求。
• 高级用法：通过事务、管道、Lua 脚本等特性，解决分布式场景、性能优化、复杂业务逻辑等问题。

实际开发中，需根据场景选择合适的用法，例如：缓存用 String/Hash，排行榜用 Sorted Set，消息队列用 Stream，独立统计用 HyperLogLog 等。