Redis Python 常用数据结构实战

目标

演示 Redis 五大核心数据结构：String、Hash、List、Set、Sorted Set，以及分布式锁。

环境准备

pip install redis
# 确保 Redis 运行
docker run -d --name redis -p 6379:6379 redis:7-alpine

完整代码

import redis
import json
import time

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

# ===== 1. String =====
r.set('app:config:timeout', '30', ex=3600)
r.setnx('app:config:version', '1.0.0')  # 不存在才设置
r.incr('app:counter:visitors')           # 原子递增
print(f"访客数: {r.get('app:counter:visitors')}")

# ===== 2. Hash（对象存储）=====
r.hset('user:1001', mapping={
    'name': 'Alice', 'email': 'alice@example.com',
    'age': 28, 'plan': 'premium'
})
print(f"用户名: {r.hget('user:1001', 'name')}")
print(f"用户信息: {r.hgetall('user:1001')}")

# ===== 3. List（消息队列）=====
r.lpush('task:queue', 'task1', 'task2', 'task3')
r.rpush('task:queue', 'task4')
task = r.blpop('task:queue', timeout=5)  # 阻塞弹出
print(f"执行任务: {task}")

# ===== 4. Set（去重集合）=====
r.sadd('user:tags:1001', 'python', 'redis', 'docker')
r.sadd('user:tags:1002', 'python', 'go', 'k8s')
common = r.sinter('user:tags:1001', 'user:tags:1002')  # 交集
print(f"共同标签: {common}")

# ===== 5. Sorted Set（排行榜）=====
scores = {'player_a': 9800, 'player_b': 8500, 'player_c': 9200, 'player_d': 7600}
r.zadd('game:leaderboard', scores)
# Top 3
top3 = r.zrevrange('game:leaderboard', 0, 2, withscores=True)
print(f"🏆 Top 3: {top3}")
# 玩家 B 的排名
rank = r.zrevrank('game:leaderboard', 'player_b')
print(f"player_b 排名: {rank + 1}")

# ===== 6. 分布式锁 =====
lock_key = 'lock:resource-a'
lock_value = f'{time.time_ns()}'

# 获取锁（NX + PX 原子操作）
acquired = r.set(lock_key, lock_value, nx=True, px=30000)
if acquired:
    try:
        print("🔒 获得锁，开始处理...")
        time.sleep(1)
    finally:
        # Lua 脚本安全释放（只释放自己的锁）
        release_script = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        else
            return 0
        end
        """
        r.eval(release_script, 1, lock_key, lock_value)
        print("🔓 释放锁")
else:
    print("❌ 锁被占用")

# ===== 7. Pipeline（批量操作）=====
pipe = r.pipeline()
for i in range(100):
    pipe.set(f'batch:key:{i}', f'value-{i}')
pipe.execute()
print("批量写入 100 条完成")

# ===== 8. Pub/Sub（发布订阅）=====
def subscriber():
    import threading
    def listen():
        pubsub = r.pubsub()
        pubsub.subscribe('channel:notifications')
        for msg in pubsub.listen():
            if msg['type'] == 'message':
                print(f"📨 收到消息: {msg['data']}")
    t = threading.Thread(target=listen, daemon=True)
    t.start()
    return t

# 在另一个进程启动 subscriber()，这里演示发布
t = subscriber()
r.publish('channel:notifications', '系统将在 10 分钟后维护')
time.sleep(0.5)

预期输出

依次输出访客数、用户信息、任务、排行榜 Top3、分布式锁状态、批量操作、Pub/Sub 消息。