Redis 使用场景：缓存、队列与限流

你已经构建了一个能正常运行的 Web 应用，但随着流量增长，你发现页面加载缓慢、数据库出现瓶颈，偶尔还会宕机。这些都是常见的扩展痛点。Redis 作为内存数据存储，可以优雅地解决许多此类问题。在本文中，我们将探讨三个实用的 Redis 使用场景：缓存、队列和限流。你将通过代码示例和最佳实践学习如何实现每一项。

为什么选择 Redis？

Redis 是一个开源的内存键值存储，以其速度和多功能性而闻名。它支持字符串、列表、集合、哈希和有序集合等数据结构，适用于各种任务。由于它将数据保存在内存中，读写速度极快——通常达到亚毫秒级。虽然它不能替代主数据库，但作为提升性能和协调的补充工具，它表现出色。

1. 使用 Redis 进行缓存

缓存是最常见的 Redis 使用场景。通过将频繁访问的数据存储在 Redis 中，可以减少主数据库的负载并加快响应速度。

如何实现缓存

1. 首先检查 Redis：当请求到达时，使用唯一键在 Redis 中查找数据。
2. 缓存未命中时：如果数据不在 Redis 中，则从数据库获取。
3. 存储到 Redis：将结果保存到 Redis 中并设置过期时间（TTL），以避免数据过时。
4. 返回数据：将数据发送给客户端。

以下是使用 redis-py 的 Python 简单示例：

import redis
import json

r = redis.Redis(host='localhost', port=6379, db=0)

def get_user(user_id):
    cache_key = f"user:{user_id}"
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)
    # 模拟数据库查询
    user = fetch_user_from_db(user_id)
    r.setex(cache_key, 3600, json.dumps(user))  # 缓存 1 小时
    return user

最佳实践：

• 使用有意义的键名，用冒号分隔，例如 user:123。
• 设置 TTL 以防止数据过时和内存膨胀。
• 考虑缓存失效策略：写穿透、写回或基于时间的过期。
• 对于复杂对象，使用 JSON 或 MessagePack 进行序列化。

2. 使用 Redis 实现队列

Redis 列表和流非常适合构建简单的消息队列。它们允许你将耗时任务（如发送邮件或处理图片）与主请求流程解耦，从而提高响应速度。

使用列表作为队列

Redis 列表支持原子性的推入和弹出操作。使用 LPUSH 将任务添加到左侧，使用 BRPOP 从右侧阻塞弹出，从而创建一个 FIFO 队列。

# 生产者：添加任务
r.lpush('job_queue', json.dumps({'task': 'send_email', 'to': 'user@example.com'}))

# 消费者：处理任务
while True:
    _, job_data = r.brpop('job_queue', timeout=30)
    job = json.loads(job_data)
    process_job(job)

注意事项：

• 可靠性：列表默认不是持久化的；如果需要，启用 AOF 或 RDB 持久化。
• 至少一次投递：使用 BRPOPLPUSH 将任务移动到处理列表，完成后移除。
• 流：对于更高级的用例（消费者组、确认机制），考虑使用 Redis Streams。

3. 使用 Redis 进行限流

限流可以保护你的 API 免受滥用，并确保公平使用。Redis 的原子操作和过期机制使其成为实现限流器的理想选择。

固定窗口限流

在固定窗口中，允许每个时间窗口内 N 个请求（例如，每小时 100 个请求）。使用 INCR 和 EXPIRE 来跟踪计数。

def is_rate_limited(user_id, limit=100, window=3600):
    key = f"rate:{user_id}"
    current = r.incr(key)
    if current == 1:
        r.expire(key, window)
    return current > limit

使用有序集合的滑动窗口提供更平滑的限流。将请求的时间戳存储在有序集合中，并移除旧条目。

def is_rate_limited_sliding(user_id, limit=100, window=3600):
    key = f"rate:{user_id}"
    now = time.time()
    pipeline = r.pipeline()
    pipeline.zremrangebyscore(key, 0, now - window)
    pipeline.zadd(key, {now: now})
    pipeline.zcard(key)
    pipeline.expire(key, window)
    _, _, count, _ = pipeline.execute()
    return count > limit

最佳实践：

• 为每个用户或 IP 使用唯一的键。
• 与中间件结合以实现清晰的集成。
• 考虑使用 Redis 模块如 redis-cell 来实现高级算法。

使用场景对比

常见问题

Redis 是持久化的吗？

Redis 通过 RDB 快照和 AOF 日志提供持久化，但它主要是一个内存存储。对于关键数据，始终要备份或使用主数据库。

我可以使用 Redis 进行会话存储吗？

是的，Redis 通常用于会话存储，因为它速度快且支持自动过期。它允许你在多个服务器之间共享会话。

如果 Redis 宕机了会发生什么？

如果 Redis 不可用，你的应用应优雅地回退到数据库或从本地缓存返回缓存数据。实现断路器以避免级联故障。

准备好优化你的 Web 应用了吗？从集成 Redis 进行缓存、队列或限流开始。对于格式化 JSON 或分析日志等快速任务，请查看我们的 JSON Formatter 和 Nginx Log Analyzer。