Redis 用例：缓存、队列与限流

Backend2026-10-13TryQuickToolBox

你已经将 Redis 加入技术栈，但只把它当作简单的键值存储使用。Redis 提供了强大的数据结构和命令，可以优雅地解决缓存、队列和限流问题。本指南将展示如何有效实现这些模式。

为什么这些场景适合用 Redis？

Redis 是一种内存数据存储，以其速度和多功能性著称。它支持字符串、哈希、列表、集合、有序集合等。其原子操作和内置过期机制使其成为缓存、队列和限流的理想选择。此外，它易于与任何语言集成。

使用 Redis 进行缓存

缓存是最常见的 Redis 用例。你将频繁访问的数据存储在 Redis 中，以减少数据库负载并加速响应。

基本的缓存旁路模式

在缓存旁路模式中，应用程序首先检查 Redis。如果数据不在那里（缓存未命中），则从数据库获取，然后存入 Redis 以备下次使用。

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）以避免数据过期。
使用有意义的键名，如 user:123:profile。
处理缓存失效：当数据变化时，更新或删除缓存条目。
考虑使用 Redis 哈希来存储对象的多个字段以节省内存。

高级缓存模式

对于高流量应用，考虑以下模式：

写穿： 同时更新缓存和数据库。
写回： 先更新缓存，然后异步更新数据库。
预刷新： 在缓存条目过期前主动刷新。

使用 Redis 实现队列

Redis 列表和流非常适合构建简单的消息队列。使用列表处理基本作业队列，使用流处理更复杂的事件处理。

使用列表的简单作业队列

使用 LPUSH 添加作业，使用 BRPOP 处理它们。BRPOP 会阻塞直到有作业可用，对工作进程来说非常高效。

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

# 消费者（工作进程）
while True:
    _, job_data = r.brpop('jobs', timeout=30)
    job = json.loads(job_data)
    process_job(job)

最佳实践：

为不同的作业类型或优先级使用单独的列表。
为失败的作业实现死信队列。
使用 BRPOPLPUSH（或 Redis 6.2+ 中的 BLMOVE）原子地将作业移动到处理列表，确保可靠性。
使用 LLEN 监控队列长度以扩展工作进程。

用于事件溯源的 Redis Streams

Redis Streams（Redis 5.0 引入）提供了更健壮的队列，支持消费者组、确认和重放。使用 XADD 添加事件，使用 XREADGROUP 消费它们。

# 添加事件
r.xadd('events', {'type': 'order_placed', 'order_id': '123'})

# 消费者组
r.xgroup_create('events', 'order_processors', id='0', mkstream=True)
while True:
    events = r.xreadgroup('order_processors', 'worker1', {'events': '>'}, count=10, block=5000)
    for event in events:
        process_event(event)
        r.xack('events', 'order_processors', event.id)

使用 Redis 实现限流

限流保护你的 API 免受滥用并确保公平使用。Redis 的原子操作使其非常适合计数请求。

固定窗口限流

为每个用户递增计数器并设置过期时间。如果计数超过限制，则阻止请求。

def is_allowed(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