Redis从基础到Stream消息队列实战指南

更新时间：2026年03月23日 09:58:03 作者：用什么都重名

Redis是一个高性能内存数据库,支持字符串、哈希、列表等数据结构,5.0版本新增Stream类型,提供消息队列功能,支持消费者组和消息确认,本文介绍Redis从基础到Stream消息队列实战指南,感兴趣的朋友跟随小编一起看看吧

一、Redis 简介

Redis（Remote Dictionary Server）是一个基于内存的键值存储系统，支持多种数据结构，常被用作缓存、消息队列和会话存储。它提供高性能读写，并支持持久化、主从复制和集群。

二、Redis 核心特点

特点	说明
内存存储	数据主要在内存中，读写延迟低（通常微秒级）
持久化	支持 RDB 快照和 AOF 日志，保证数据不丢失
多种数据结构	String、List、Hash、Set、Sorted Set、Stream、Bitmap 等
单线程模型	命令串行执行，避免锁竞争，保证原子性
主从复制	支持读写分离和高可用
发布订阅	支持 Pub/Sub 和 Stream 消息队列
事务	支持 MULTI/EXEC 事务
Lua 脚本	支持 Lua 脚本，保证原子性
集群	支持 Redis Cluster 水平扩展

三、Redis 常用数据结构

3.1 String（字符串）

SET key value          # 设置
GET key                # 获取
INCR key               # 自增
EXPIRE key seconds     # 设置过期时间

3.2 Hash（哈希）

HSET user:1 name "张三" age 25
HGET user:1 name
HGETALL user:1

3.3 List（列表）

LPUSH queue task1      # 左侧入队
RPOP queue             # 右侧出队
LRANGE queue 0 -1      # 范围查询

3.4 Set（集合）

SADD tags "redis" "cache"
SMEMBERS tags
SISMEMBER tags "redis"

3.5 Sorted Set（有序集合）

ZADD rank 100 "user1" 95 "user2"
ZRANGE rank 0 -1 WITHSCORES
ZREVRANK rank "user1"

四、Redis Stream：消息队列

Redis 5.0 引入 Stream，用于实现消息队列，支持：

消息持久化

消费者组

消息确认（ACK）

历史消息回溯

4.1 基本命令

# 添加消息
XADD mystream * field1 value1 field2 value2
# 读取消息（从头）
XREAD COUNT 10 STREAMS mystream 0
# 读取最新消息（阻塞）
XREAD BLOCK 5000 STREAMS mystream $

4.2 消费者组

# 创建消费者组
XGROUP CREATE mystream mygroup 0 MKSTREAM
# 消费消息
XREADGROUP GROUP mygroup consumer1 COUNT 1 STREAMS mystream >
# 确认消息
XACK mystream mygroup message-id
# 查看待确认消息
XPENDING mystream mygroup

4.3 Stream 与 List 的对比

特性	List	Stream
消息持久化	有	有
消费者组	无	有
消息确认	无	有
历史回溯	有限	支持
适用场景	简单队列	可靠消息队列

五、Redis 在项目中的典型用法

以银行场景下的 SOP 合规检测为例，Redis Stream 用于任务分发和结果回传。

5.1 架构示意

后端/前端 → XADD 推送任务 → Redis Stream
↓
sop_engine 监听 XREADGROUP
↓
处理完成后 XADD 推送结果
↓
后端消费 message:xxx:results

5.2 MAF 营销分析任务流

# 1. 推送 MAF 任务
XADD message:maf:tasks * task_id "maf_001" \
  conversation_id "maf_001" \
  conversation_json_path "bucket/path/conversation.json"
# 2. 引擎监听 message:maf:tasks，消费并分析
# 3. 引擎完成后写入结果流
XADD message:maf:results * task_id "maf_001" status "completed" \
  result_path "maf-results-bucket/maf_001/analysis_result.json"

5.3 消费者组带来的好处

多个 worker 共同消费，实现负载均衡

每条消息只被组内一个消费者处理

支持 ACK，失败可重试

支持 PEL（Pending Entries List）查看未确认消息

5.4 Redis Stream 请求分发规则

如果两个项目使用了同一套Redis配置，那么无法保证Redis Stream 请求会由谁处理，取决于 Redis 消费者组的分发。

两个容器都连同一个 Redis，监听同一个 Stream（如 message:maf:tasks），且通常在同一消费者组（如 maf_analyze_group）里：

Redis 会把消息分发给组内的消费者

每条消息只会被组内一个消费者处理

具体是容器 1 还是容器 2，由 Redis 的分发策略决定，无法指定

因此：

有时是A容器处理

有时是B容器处理

测试时无法稳定地“只让某个容器”处理请求

可能带来的问题：

场景	说明
代码版本混用	同一批任务，部分由旧代码处理，部分由新代码处理
测试不可控	无法保证测试请求一定打到新容器
任务重复	若消费者组配置不当，可能出现同一条消息被多个消费者处理

如何保证只使用新代码：

方案 1：停掉旧容器

# 停掉旧容器后再部署新容器
docker stop <old_container_id>
docker run ...  # 启动新容器

方案 2：用不同的 Stream 做测试

新容器监听不同的 Stream，例如 message:maf:tasks_test

测试时往 message:maf:tasks_test 发消息

需要改配置或环境变量，让新容器使用 message:maf:tasks_test

方案 3：用不同的消费者组

新容器使用不同的消费者组名，例如 maf_analyze_group_v2

两个组都会收到同一条消息，各自处理一次

适合做 A/B 或灰度，但会产生重复处理，需要业务上能接受

方案 4：只保留一个容器

部署新容器前先停掉旧容器

或使用滚动更新，保证同一时间只有一个版本在跑

六、Redis 常用配置

# 绑定地址
bind 0.0.0.0
# 端口
port 6379
# 密码
requirepass your_password
# 最大内存
maxmemory 2gb
maxmemory-policy allkeys-lru
# 持久化
save 900 1
save 300 10
save 60 10000
appendonly yes

七、Python 操作 Redis

7.1 Linux安装

# Ubuntu/Debian
sudo apt update
sudo apt install redis-server
# CentOS/RHEL
sudo yum install redis
# 启动服务
sudo systemctl start redis
sudo systemctl enable redis
# 验证
redis-cli ping
# 返回 PONG 表示成功

7.2 Docker安装

docker run -d --name redis -p 6379:6379 redis:latest
# 带密码
docker run -d --name redis -p 6379:6379 redis redis-server --requirepass yourpassword

7.3 连接Redis

# 本地连接
redis-cli
# 带密码连接
redis-cli -a yourpassword
# 远程连接
redis-cli -h host -p 6379 -a password

7.4 基本使用

import redis
r = redis.Redis(host='localhost', port=6379, db=0, password='')
# String
r.set('name', 'Redis')
print(r.get('name'))
# Stream 添加消息
r.xadd('mystream', {'task_id': '001', 'data': 'hello'})
# Stream 消费
messages = r.xreadgroup('mygroup', 'consumer1', {'mystream': '>'}, count=1)
for stream, msgs in messages:
    for msg_id, data in msgs:
        print(msg_id, data)
        r.xack('mystream', 'mygroup', msg_id)

八、Redis 使用建议

合理设置 maxmemory 和淘汰策略，避免 OOM。
生产环境开启 requirepass 和访问控制。
使用连接池，减少连接开销。
对热点 key 做拆分或本地缓存，减轻压力。
使用 Pipeline 批量执行命令，降低网络往返。
Stream 场景下注意消费者组和 ACK，避免消息堆积和重复消费。

总结

Redis 适合做缓存、会话存储和消息队列。Stream 在需要可靠消费、负载均衡和消息确认的场景中，比简单 List 更合适。结合具体业务（如 MAF 任务流），可以设计出清晰、可扩展的异步处理架构。

到此这篇关于Redis从基础到Stream消息队列实战指南的文章就介绍到这了,更多相关redis stream 消息队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

Redis特殊类型数据结构Bitmap、HyperLogLog、GEO的使用及场景分析
文章介绍了Redis的三种特殊数据类型:Bitmap、HyperLogLog和GEO,分别用于不同的场景,Bitmap适合存储大量二进制数据,HyperLogLog在大数据场景下能高效统计基数,而GEO则用于地理空间信息的管理和查询,本文介绍的非常详细,感兴趣的朋友跟随小编一起通过本文学习吧
2025-12-12
Redis如何使用zset处理排行榜和计数问题
Redis的ZSET数据结构非常适合处理排行榜和计数问题,它可以在高并发的点赞业务中高效地管理点赞的排名,并且由于ZSET的排序特性,可以轻松实现根据点赞数实时排序的功能
2025-02-02
Redis持久化解读
Redis是一种内存级数据库,提供高速读写性能,但数据易失,它支持三种持久化方式：RDB（快照持久化）、AOF（追加文件持久化）和混合持久化,RDB通过快照将数据保存到磁盘,AOF记录所有写操作命令,混合持久化结合两者优点
2025-01-01
Redis实现优惠券限一单限制详解
这篇文章主要介绍了Redis解决优惠券秒杀应用案例,本文先讲了抢购问题，指出其中会出现的多线程问题，提出解决方案采用悲观锁和乐观锁两种方式进行实现，然后发现在抢购过程中容易出现一人多单现象，需要的朋友可以参考下
2022-12-12
批量导入txt数据到的redis过程
用户通过将Redis命令逐行写入txt文件,利用管道模式运行客户端,成功执行批量删除以"Product*"匹配的Key操作,提高了数据清理效率
2025-08-08
Redis主从配置和底层实现原理解析(实战记录)
今天给大家分享Redis主从配置和底层实现原理解析，本文通过实战项目给大家源码解析，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友参考下吧
2021-06-06
使用Redis实现请求限制与速率限制
API速率限制（Rate Limiting）是控制用户访问API的请求速率的一种机制,防止系统被过多请求淹没,下面我们来看看如何使用Redis和FastAPI实现请求限制与速率控制吧
2025-04-04
Centos7下Redis3.2.8最新版本安装教程
这篇文章主要为大家详细介绍了Centos7下Redis3.2.8最新版本的安装教程，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2017-04-04
Redis并发问题解决方案
在当前的互联网环境中,高并发业务场景十分常见,本文就来介绍一下Redis并发问题解决方案,具有一定的参考价值,感兴趣的可以了解一下
2023-11-11
Redis操作相关命令之查看、停止、启动命令
这篇文章主要介绍了Redis操作相关命令之查看、停止、启动命令,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2023-09-09