当前位置：首页 > news >正文

Redis 在互联网高并发场景下的应用--个人总结

news 2025/9/8 8:26:27

在现代互联网系统中，高并发已经成为常态。无论是电商的秒杀场景、社交平台的热点推荐，还是支付接口的风控，系统需要同时应对成千上万的请求。这时候，Redis 作为一个高性能的内存数据库，凭借其极快的读写速度和丰富的数据结构，成为解决高并发问题的利器。

1. 缓存热点数据 —— 缓解数据库压力

问题背景：数据库在高并发场景下容易成为瓶颈，特别是热点数据的访问会导致 QPS 瞬间飙升。
Redis 解决方案：

使用 String 或 Hash 缓存热点数据。
采用 Cache Aside 模式：先查缓存，缓存未命中再查数据库并回填。
避免常见问题：
- 缓存穿透：使用布隆过滤器或缓存空对象。
- 缓存雪崩：给缓存设置随机过期时间或提前预热。
- 缓存击穿：对热点 key 加互斥锁，避免同时穿透到数据库。

应用场景：用户信息、商品详情页、首页推荐数据等。

示例代码：

import redisr = redis.Redis(host='localhost', port=6379, db=0)def get_data_from_db(key):print("查询数据库")return {"id": key, "value": f"data_{key}"}def get_data(key):cache_key = f"cache:{key}"val = r.get(cache_key)if val:print("命中缓存")return eval(val)data = get_data_from_db(key)r.setex(cache_key, 60, str(data))return dataprint(get_data(1))
print(get_data(1))

2. 限流 —— 防止接口过载

问题背景：突发流量可能会压垮下游服务，例如秒杀、登录、支付接口。
Redis 解决方案：

使用 String 作为计数器：INCR + EXPIRE 实现固定时间窗口限流。
使用 Sorted Set 存储时间戳：实现滑动窗口限流，更加精细化。
可以配合 Lua 脚本保证原子性操作。

应用场景：API 调用限流、防止刷单、短信/验证码接口保护。

示例代码（滑动窗口限流）：

import time
import redisr = redis.Redis(host='localhost', port=6379, db=0)def is_allowed_sliding_window(user_id, action, period=60, max_count=5):key = f"rate_limit:{user_id}:{action}"now = time.time()pipeline = r.pipeline()pipeline.zadd(key, {now: now})pipeline.zremrangebyscore(key, 0, now - period)pipeline.zcard(key)pipeline.expire(key, period)_, _, count, _ = pipeline.execute()return count <= max_countfor i in range(10):if is_allowed_sliding_window("u1001", "login"):print(i, "允许")else:print(i, "拒绝")time.sleep(0.5)

或者使用 String 作为计数器

import redis
import timer = redis.Redis(host='localhost', port=6379, db=0)def is_allowed(user_id, action, period=60, max_count=5):"""在 period 时间内最多允许 max_count 次操作"""key = f"rate_limit:{user_id}:{action}"count = r.incr(key)if count == 1:r.expire(key, period)  # 设置过期时间if count > max_count:return Falsereturn True# 测试
for i in range(10):if is_allowed("u1001", "login"):print(i, "允许")else:print(i, "拒绝")time.sleep(0.5)

3. 分布式队列 —— 异步削峰

问题背景：大量请求直接打到数据库或下游服务，可能会瞬间撑爆系统。
Redis 解决方案：

使用 List（LPUSH + BRPOP）作为简单任务队列。
使用 Stream 实现类似 Kafka 的消息队列：支持消费组、消息确认和持久化。
常见用途：削峰填谷，将请求转为异步任务，让系统有更多缓冲时间。

应用场景：订单处理、日志收集、消息通知、秒杀排队。

示例代码（Stream 队列）：

import redisr = redis.Redis(host='localhost', port=6379, db=0)
stream_key = "order_stream"# 生产者
def produce():for i in range(5):r.xadd(stream_key, {"order": f"order_{i}"})print("生产订单:", f"order_{i}")# 消费者
def consume():last_id = "0-0"while True:messages = r.xread({stream_key: last_id}, block=2000, count=2)if not messages:breakfor _, msgs in messages:for msg_id, data in msgs:print("消费订单:", msg_id, data)last_id = msg_idproduce()
consume()

4. 分布式锁 —— 保证数据一致性

问题背景：在分布式环境中，多实例同时修改同一资源，可能造成数据冲突。
Redis 解决方案：

使用 String 类型：
- SET key value NX PX timeout 实现加锁，避免重复设置。
- 校验 value 防止误删锁。
高级方案：使用 RedLock 算法 提升可靠性，避免单点故障。

应用场景：库存扣减、任务调度、并发下的幂等操作。

示例代码：

import redis
import uuid
import timer = redis.Redis(host='localhost', port=6379, db=0)def acquire_lock(lock_key, expire=5):lock_id = str(uuid.uuid4())result = r.set(lock_key, lock_id, nx=True, ex=expire)return lock_id if result else Nonedef release_lock(lock_key, lock_id):script = """if redis.call('get', KEYS[1]) == ARGV[1] thenreturn redis.call('del', KEYS[1])elsereturn 0end"""return r.eval(script, 1, lock_key, lock_id)lock_id = acquire_lock("my_lock", expire=3)
if lock_id:print("获得锁，执行业务逻辑...")time.sleep(2)print("释放锁:", release_lock("my_lock", lock_id))
else:print("未获得锁")

5. 实时统计与排行榜

问题背景：高并发下，需要对访问量、点赞数、排名进行实时计算。
Redis 解决方案：

使用 INCR / HINCRBY 做计数器。
使用 Sorted Set 实现排行榜，支持按分数排序。

应用场景：网站 PV/UV 统计、热搜榜、积分排名、点赞排行榜。

示例代码：

import redisr = redis.Redis(host='localhost', port=6379, db=0)# 统计 PV
r.incr("page_view")
print("页面访问量:", r.get("page_view").decode())# 排行榜
leaderboard = "game_rank"
r.zincrby(leaderboard, 10, "user1")
r.zincrby(leaderboard, 20, "user2")print("排行榜:", r.zrevrange(leaderboard, 0, -1, withscores=True))