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

Backend2026-10-04TryQuickToolBox

你已经构建了一个正在获得关注的 Web 应用，但现在每次页面加载都会访问数据库，后台任务堆积如山，还有少数用户在猛击你的 API。你需要一个快速的内存数据存储来应对这些挑战。Redis 是缓存、队列和限流的首选解决方案。在本文中，我们将通过实际示例和最佳实践来探讨这三个核心使用场景。

为什么选择 Redis 进行缓存、队列和限流？

Redis 是一个以速度和多功能性著称的内存键值存储。它支持丰富的数据结构，如字符串、列表、集合、有序集合和哈希，使其成为各种任务的理想选择。其原子操作和发布/订阅功能进一步扩展了其实用性。无论你是要减少数据库负载、解耦服务还是保护 API，Redis 都提供了一个简单而强大的工具包。

1. 使用 Redis 进行缓存

缓存是 Redis 最常见的用例。通过将频繁访问的数据存储在内存中，可以减少延迟和数据库负载。Redis 的 TTL（生存时间）功能会自动使过期数据失效。

如何实现缓存

检查缓存：在查询数据库之前，检查数据是否存在于 Redis 中。
缓存未命中时：从数据库获取数据，并将结果存储到 Redis 中，设置过期时间。
缓存命中时：直接返回缓存数据。

以下是使用 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

缓存最佳实践

设置适当的 TTL：平衡新鲜度和命中率。对于易变数据使用较短的 TTL，对于静态内容使用较长的 TTL。
使用缓存失效：当数据发生变化时，删除或更新缓存键，以避免读取到过期数据。
处理缓存雪崩：使用锁或概率性提前过期，以防止在缓存未命中时大量请求同时访问数据库。
监控命中率：力求高命中率（例如 >80%），以确保缓存有效。

2. 使用 Redis 实现队列

Redis 列表和有序集合可实现简单而强大的消息队列。它们非常适合后台作业、任务队列和事件处理。与专用消息代理不同，Redis 队列轻量且易于设置。

实现基本队列

使用 LPUSH 添加作业，使用 BRPOP 以阻塞方式消费它们：

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

# 消费者
while True:
    _, job = r.brpop('job_queue', timeout=30)
    if job:
        process_job(json.loads(job))

使用有序集合的高级队列

对于延迟作业或优先级队列，使用以时间戳作为分数的有序集合。消费者获取分数小于当前时间的作业：

# 添加延迟作业（延迟 60 秒）
import time
score = time.time() + 60
r.zadd('delayed_queue', {json.dumps({'task': 'send_reminder'}): score})

# 消费者获取到期作业
now = time.time()
jobs = r.zrangebyscore('delayed_queue', 0, now, start=0, num=1)
if jobs:
    job = jobs[0]
    if r.zrem('delayed_queue', job):  # 原子性移除
        process_job(json.loads(job))

队列最佳实践

使用可靠的队列模式：结合 BRPOPLPUSH 或 BLMOVE 与处理列表，确保作业在失败时不会丢失。
处理重复项：使用唯一作业 ID 和集合来防止重复处理。
监控队列长度：跟踪队列大小，以便在积压时进行扩展。
设置合理的超时：为阻塞操作使用超时，以避免无限期等待。

3. 使用 Redis 进行限流

限流保护你的 API 免受滥用，并确保公平使用。Redis 的原子递增和过期命令使其成为实现限流器的理想选择。

固定窗口限流器

使用 INCR 和 EXPIRE 实现一个简单限流器：

def is_rate_limited(user_id, limit=100, window=60):
    key = f"rate_limit:{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=60):
    key = f"rate_limit_sliding:{user_id}"
    now = time.time()
    # 移除窗口外的旧请求
    r.zremrangebyscore(key, 0, now - window)
    # 添加当前请求
    r.zadd(key, {str(now): now})
    # 统计窗口内的请求数
    count = r.zcard(key)
    r.expire(key, window)  # 设置过期时间以清理
    return count > limit

限流最佳实践

选择合适的算法：固定窗口简单但可能导致突发流量；滑动窗口更平滑但更复杂。
使用 Redis 集群：对于分布式系统，确保限流键分布均匀，或使用中心化存储。
返回适当的 HTTP 状态码：当限流触发时，返回 429 Too Many Requests。
记录限流事件：监控被限流的请求，以调整限制。

结论

Redis 是一个多功能工具，可以显著提升 Web 应用的性能、可靠性和安全性。通过为缓存、队列和限流实施这些模式，你可以构建出能优雅处理扩展的稳健系统。记住要监控、调整并遵循最佳实践，以充分利用 Redis。