Hello World

目标

通过 redis-cli 和 Python redis 库，完成 Redis 五种核心数据结构的操作，体验缓存加速效果。

环境准备

redis-cli ping  # 确保返回 PONG
pip install redis

第一步：redis-cli 方式

redis-cli

# === String（字符串）===
SET username "alice"
GET username
INCR page_views          # 原子自增计数器
INCRBY page_views 10     # 一次性 +10
EXPIRE page_views 3600   # 1小时后过期

# === Hash（哈希 - 对象存储）===
HSET user:1001 name "张三" age "21" email "zs@example.com"
HGET user:1001 name
HGETALL user:1001
HINCRBY user:1001 age 1  # 年龄 +1

# === List（列表 - 消息队列）===
LPUSH queue:tasks "task1" "task2" "task3"
RPOP queue:tasks          # 取出最后一个
LLEN queue:tasks
LRANGE queue:tasks 0 -1   # 查看全部

# === Set（集合 - 去重/交并差）===
SADD skills:alice "Python" "Redis" "MySQL"
SADD skills:bob "Python" "MongoDB" "Redis"
SINTER skills:alice skills:bob   # 共同技能
SUNION skills:alice skills:bob   # 所有技能

# === Sorted Set（有序集合 - 排行榜）===
ZADD leaderboard 92 "Alice" 88 "Bob" 95 "Charlie" 87 "David"
ZRANGE leaderboard 0 -1 WITHSCORES    # 全部排名
ZREVRANGE leaderboard 0 2 WITHSCORES  # Top 3
ZINCRBY leaderboard 5 "Alice"         # Alice 加分

第二步：Python 操作

import redis
import time

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

# === 缓存模式 ===
def get_user(user_id):
    """先从缓存取，未命中再查'数据库'"""
    cache_key = f"user:{user_id}"
    cached = r.hgetall(cache_key)
    if cached:
        print(f"[缓存命中] {cached}")
        return cached

    # 模拟数据库查询
    print(f"[数据库查询] user:{user_id}")
    time.sleep(0.1)  # 模拟慢查询
    user_data = {"name": "张三", "age": 21, "email": f"user{user_id}@example.com"}

    # 写入缓存（30秒过期）
    r.hset(cache_key, mapping=user_data)
    r.expire(cache_key, 30)
    return user_data

print(get_user(1001))  # 第一次：数据库
print(get_user(1001))  # 第二次：缓存命中

# === 分布式锁 ===
def do_critical_task():
    lock_key = "lock:task"
    # 尝试获取锁（SET NX + EX = 原子操作）
    acquired = r.set(lock_key, "locked", nx=True, ex=10)
    if not acquired:
        print("任务正在执行中，跳过")
        return

    try:
        print("执行关键任务...")
        time.sleep(1)
    finally:
        r.delete(lock_key)

# === 发布订阅 ===
def subscriber():
    pubsub = r.pubsub()
    pubsub.subscribe("channel:news")
    for msg in pubsub.listen():
        if msg['type'] == 'message':
            print(f"收到消息: {msg['data']}")

# 另一个进程发布：
# r.publish("channel:news", "Hello Redis!")

第三步：缓存加速效果对比

import timeit

def without_cache():
    time.sleep(0.05)  # 模拟每次 DB 查询
    return {"data": "result"}

def with_cache():
    cached = r.get("expensive_data")
    if cached:
        return cached
    data = without_cache()
    r.setex("expensive_data", 60, str(data))
    return data

# 预热缓存
with_cache()

t1 = timeit.timeit(without_cache, number=100)
t2 = timeit.timeit(with_cache, number=100)
print(f"无缓存 100次: {t1:.2f}s")
print(f"有缓存 100次: {t2:.2f}s  (加速 {t1/t2:.0f}x)")

预期输出

# redis-cli
127.0.0.1:6379> ZREVRANGE leaderboard 0 2 WITHSCORES
1) "Charlie" 2) "95"
3) "Alice"   4) "92"
5) "Bob"     6) "88"

# Python
[数据库查询] user:1001
{'name': '张三', 'age': 21, 'email': 'user1001@example.com'}
[缓存命中] {'name': '张三', 'age': 21, 'email': 'user1001@example.com'}

无缓存 100次: 5.10s
有缓存 100次: 0.05s  (加速 102x)