Redis 常见数据类型的使用场景

user_id = 1001
cached_user = redis_client.get(f"user:{user_id}")
if cached_user:
    return json.loads(cached_user)
else:
    user = db.query(User).get(user_id)  # 从数据库查询
    redis_client.setex(f"user:{user_id}", 3600, json.dumps(user))  # 缓存 1 小时
    return user

def generate_short_code(url):
    """生成短链代码（如 6 位随机字符串或基于 URL 的哈希）"""
    # 简单示例：随机 6 位字符
    characters = string.ascii_letters + string.digits
    return ''.join(random.choice(characters) for _ in range(6))

def create_short_url(long_url, expire_seconds=2592000):
    """创建短链并存储"""
    short_code = generate_short_code(long_url)
    # 确保 short_code 唯一，实际中可能需要重试或哈希
    while redis_client.exists(f"short:{short_code}"):
        short_code = generate_short_code(long_url)
    redis_client.setex(f"short:{short_code}", expire_seconds, long_url)
    return short_code

def get_long_url(short_code):
    """根据短链获取原始网址"""
    long_url = redis_client.get(f"short:{short_code}")
    return long_url.decode() if long_url else None

HMSET user:1001 id "1001" name "Alice" email "[email protected]"
HSET user:1001 name "Bob"  # 只修改 name 字段
HGET user:1001 name       # 只获取 name 字段

# 生产者：订单支付后推送任务
def add_notification_task(order_id, user_email):
    task = f"send_email:{order_id}:{user_email}"
    r.lpush("notification_queue", task)  # 推入队列头部
    print(f"Task added for order {order_id}")

# 消费者：邮件服务处理任务
def process_notification():
    while True:
        # 阻塞等待任务，最多等 10 秒
        task = r.brpop("notification_queue", timeout=10)
        if task:
            _, task_data = task  # task 是 (key, value) 元组
            print(f"Processing: {task_data.decode()}")
            # 解析任务并发送邮件（伪代码）
            order_id, email = task_data.decode().split(":")[1:]
            send_email(email, f"Order {order_id} confirmed!")

# 记录用户编辑操作
def record_edit(user_id, operation):
    key = f"edit_history:{user_id}"
    r.lpush(key, operation)  # 压入操作
    r.ltrim(key, 0, 99)  # 限制最多 100 条历史
    print(f"Recorded: {operation}")

# 撤销操作
def undo_edit(user_id):
    key = f"edit_history:{user_id}"
    operation = r.lpop(key)  # 弹出最近操作
    if operation:
        print(f"Undoing: {operation.decode()}")
        # 执行反向操作（伪代码）
        reverse_operation(operation.decode())
    else:
        print("No operations to undo")

# 测试
record_edit("user123", "insert:text:hello")
record_edit("user123", "delete:char:5")
undo_edit("user123")  # 输出: Undoing: delete:char:5

# 用户发布动态
def post_update(user_id, content):
    post_id = int(time.time() * 1000)  # 用时间戳作为 ID
    post_data = f"{post_id}:{user_id}:{content}"
    r.lpush("timeline:global", post_data)  # 推入全局时间线
    r.ltrim("timeline:global", 0, 999)  # 限制 1000 条
    print(f"Posted: {content}")

# 获取最新动态
def get_recent_posts(limit=10):
    posts = r.lrange("timeline:global", 0, limit - 1)  # 取最新 10 条
    return [p.decode() for p in posts]

# 测试
post_update("user123", "Hello, world!")
post_update("user456", "Nice day!")
recent_posts = get_recent_posts()
for post in recent_posts:
    print(post)