脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 数据库 → Redis → redis stream 消息队列

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 的对比

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

五、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 使用建议

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

总结

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

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

您可能感兴趣的文章:

相关文章

  • Redis sentinel节点如何修改密码

    Redis sentinel节点如何修改密码

    这篇文章主要介绍了Redis sentinel节点如何修改密码问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
    2024-01-01
  • 详解如何清理redis集群的所有数据

    详解如何清理redis集群的所有数据

    这篇文章主要介绍了详解如何清理redis集群的所有数据，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2021-02-02
  • Redis的RDB持久化使用及说明

    Redis的RDB持久化使用及说明

    这篇文章主要介绍了Redis的RDB持久化使用及说明,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
    2025-12-12
  • 用Redis实现微博关注关系

    用Redis实现微博关注关系

    在微博中，每一个用户都会有一个关注列表，一个粉丝列表。用户可以查看自己的关注，粉丝列表，也可以查看别人的关注，粉丝列表。并且，要展示列表里每个人与当前查看者的关注状态。
    2015-09-09
  • 使用Redis实现实时排行榜功能

    使用Redis实现实时排行榜功能

    排行榜功能是一个很普遍的需求。使用 Redis 中有序集合的特性来实现排行榜是又好又快的选择。接下来通过本文给大家介绍使用Redis实现实时排行榜功能,需要的朋友可以参考下
    2021-07-07
  • Windows下安装Redis服务的图文教程

    Windows下安装Redis服务的图文教程

    Redis是有名的NoSql数据库，一般Linux都会默认支持。但在Windows环境中，可能需要手动安装设置才能有效使用。下面通过本文给大家介绍Windows下安装Redis服务的图文教程，感兴趣的朋友一起看看吧
    2018-08-08
  • Redis如何使用HyperLogLog的实现

    Redis如何使用HyperLogLog的实现

    本文主要介绍了Redis如何使用HyperLogLog的实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2022-06-06
  • Redis 热 key 和大 key 问题小结

    Redis 热 key 和大 key 问题小结

    这篇文章主要介绍了Redis 热 key 和大 key 问题小结,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧
    2025-04-04
  • 通过prometheus监控redis实时运行状态的操作方法

    通过prometheus监控redis实时运行状态的操作方法

    本文详细介绍了如何通过Prometheus监控Redis的运行状态,包括安装配置Redis、Redis Exporter以及Prometheus,配置Prometheus监控Redis指标,以及常见的Redis指标和告警规则,需要的朋友可以参考下
    2025-02-02
  • Redis 集群模式优化和备份策略

    Redis 集群模式优化和备份策略

    本文主要介绍了Linux系统和Redis集群模式的优化和备份策略,在Linux系统优化方面,重点介绍了内存管理、文件句柄、网络/TCP调优、CPU调度和频率、I/O子系统等参数的调整建议,感兴趣的朋友跟随小编一起看看吧
    2025-11-11

最新评论