Redis 基础教程

下面给你整理一份最完整、最易懂、可直接上手的 Python redis 库教程（基于最新版 redis-py 5.x），包含：连接、字符串、列表、集合、有序集合、哈希、过期、管道、发布订阅、连接池、JSON 模块、实际项目建议等。

1. 安装

pip install redis

2. 连接 Redis

import redis

r = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True  # 自动解码字符串为 str
)

3. 字符串（String）

设置值

r.set("name", "alice")

获取值

print(r.get("name"))   # alice

设置过期（秒）

r.set("token", "abc123", ex=10)  # 10秒后过期

原子递增 / 递减

r.incr("counter")      # +1
r.incrby("counter", 5) # +5
r.decr("counter")      # -1

4. 列表（List）

添加元素

r.rpush("numbers", "1", "2", "3")  # 右推入
r.lpush("numbers", "0")            # 左推入

查看列表

r.lrange("numbers", 0, -1)

弹出元素

r.lpop("numbers")  # 左弹
r.rpop("numbers")  # 右弹

检查元素是否存在

Redis 本身无“列表 contains”命令，要手动查：

items = r.lrange("numbers", 0, -1)
if "4" not in items:
    r.rpush("numbers", "4")

5. 集合（Set）

添加元素

r.sadd("tags", "python", "redis")

查看成员

r.smembers("tags")

判断是否存在

r.sismember("tags", "python")  # True

6. 有序集合（Sorted Set）

添加

r.zadd("scores", {"alice": 100, "bob": 95})

获取指定范围

r.zrange("scores", 0, -1, withscores=True)

7. 哈希（Hash）

设置 / 获取字段

r.hset("user:1", mapping={"name": "Alice", "age": 20})
r.hget("user:1", "name")

获取全部

r.hgetall("user:1")

8. 键与过期

删除键

r.delete("mykey")

设置过期

r.expire("session", 60)  # 60秒

查看剩余 TTL

r.ttl("session")

9. 管道（Pipeline）—— 批量执行，减少网络开销

pipe = r.pipeline()
pipe.set("a", 1)
pipe.incr("a")
pipe.get("a")
result = pipe.execute()
print(result)

10. 发布 / 订阅（Pub/Sub）

发布消息

r.publish("news", "hello redis")

订阅消息

pubsub = r.pubsub()
pubsub.subscribe("news")

for msg in pubsub.listen():
    print(msg)

11. 连接池（推荐用于高并发）

pool = redis.ConnectionPool(host='localhost', port=6379, db=0)

r = redis.Redis(connection_pool=pool)

12. Redis JSON（如果服务器安装了 RedisJSON 模块）

安装：

pip install redis

使用：

from redis import Redis
from redis.commands.json.path import Path

r = Redis()

r.json().set("user:1", Path.root_path(), {"name": "Alice", "age": 20})

print(r.json().get("user:1"))

13. 实战案例：数字每秒递增

你刚刚问的：

import redis, time

r = redis.Redis(decode_responses=True)

key = "detail_latest_number"
r.set(key, 100)

while True:
    r.incr(key)
    print("当前值：", r.get(key))
    time.sleep(1)

使用 incr 是原子操作，比手动 get+set 更安全。

14. 实战案例：列表去重插入

key = "cls_detail_numbers"

if "4" not in r.lrange(key, 0, -1):
    r.rpush(key, "4")

15. 清空列表

r.delete("cls_detail_numbers")

16. 项目中常用最佳实践

使用 decode_responses=True，避免 bytes
使用连接池提高性能
使用 incr/incrby 做计数器
使用 TTL 防止键无限增长
保存结构化数据尽量使用 hash 或 RedisJSON
高并发快速队列使用 list / stream

Python 生产级用法（超级关键🔥）

使用连接池

import redis

pool = redis.ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,
    max_connections=200
)

r = redis.Redis(connection_pool=pool)

绝不要使用 get+set 自增

使用原子操作：

r.incr("counter")

使用 Pipeline 批量执行

pipe = r.pipeline()
pipe.incr("views")
pipe.sadd("users", "martin")
result = pipe.execute()

使用 SCAN 而非 KEYS

for key in r.scan_iter("*"):
    print(key)

Redis 作为锁（分布式锁）

with r.lock("my_lock", timeout=10):
    # 业务代码