一、Redis 簡介

Redis（Remote Dictionary Server）是一個基于內(nèi)存的鍵值存儲系統(tǒng)，支持多種數(shù)據(jù)結(jié)構(gòu)，常被用作緩存、消息隊(duì)列和會話存儲。它提供高性能讀寫，并支持持久化、主從復(fù)制和集群。

二、Redis 核心特點(diǎn)

三、Redis 常用數(shù)據(jù)結(jié)構(gòu)

3.1 String（字符串）

SET key value          # 設(shè)置
GET key                # 獲取
INCR key               # 自增
EXPIRE key seconds     # 設(shè)置過期時間

3.2 Hash（哈希）

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

3.3 List（列表）

LPUSH queue task1      # 左側(cè)入隊(duì)
RPOP queue             # 右側(cè)出隊(duì)
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：消息隊(duì)列

Redis 5.0 引入 Stream，用于實(shí)現(xiàn)消息隊(duì)列，支持：

• 消息持久化

• 消費(fèi)者組

• 消息確認(rèn)（ACK）

• 歷史消息回溯

4.1 基本命令

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

4.2 消費(fèi)者組

# 創(chuàng)建消費(fèi)者組
XGROUP CREATE mystream mygroup 0 MKSTREAM
# 消費(fèi)消息
XREADGROUP GROUP mygroup consumer1 COUNT 1 STREAMS mystream >
# 確認(rèn)消息
XACK mystream mygroup message-id
# 查看待確認(rèn)消息
XPENDING mystream mygroup

4.3 Stream 與 List 的對比

五、Redis 在項(xiàng)目中的典型用法

以銀行場景下的 SOP 合規(guī)檢測為例，Redis Stream 用于任務(wù)分發(fā)和結(jié)果回傳。

5.1 架構(gòu)示意

后端/前端 → XADD 推送任務(wù) → Redis Stream
                              ↓
                    sop_engine 監(jiān)聽 XREADGROUP
                              ↓
                    處理完成后 XADD 推送結(jié)果
                              ↓
                    后端消費(fèi) message:xxx:results

5.2 MAF 營銷分析任務(wù)流

# 1. 推送 MAF 任務(wù)
XADD message:maf:tasks * task_id "maf_001" \
  conversation_id "maf_001" \
  conversation_json_path "bucket/path/conversation.json"
# 2. 引擎監(jiān)聽 message:maf:tasks，消費(fèi)并分析
# 3. 引擎完成后寫入結(jié)果流
XADD message:maf:results * task_id "maf_001" status "completed" \
  result_path "maf-results-bucket/maf_001/analysis_result.json"

5.3 消費(fèi)者組帶來的好處

• 多個 worker 共同消費(fèi)，實(shí)現(xiàn)負(fù)載均衡

• 每條消息只被組內(nèi)一個消費(fèi)者處理

• 支持 ACK，失敗可重試

• 支持 PEL（Pending Entries List）查看未確認(rèn)消息

5.4 Redis Stream 請求分發(fā)規(guī)則

如果兩個項(xiàng)目使用了同一套Redis配置，那么無法保證Redis Stream 請求會由誰處理，取決于 Redis 消費(fèi)者組的分發(fā)。

兩個容器都連同一個 Redis，監(jiān)聽同一個 Stream（如 message:maf:tasks），且通常在同一消費(fèi)者組（如 maf_analyze_group）里：

• Redis 會把消息分發(fā)給組內(nèi)的消費(fèi)者

• 每條消息只會被組內(nèi)一個消費(fèi)者處理

• 具體是容器 1 還是容器 2，由 Redis 的分發(fā)策略決定，無法指定

因此：

• 有時是A容器處理

• 有時是B容器處理

• 測試時無法穩(wěn)定地“只讓某個容器”處理請求

可能帶來的問題：

如何保證只使用新代碼：

方案 1：停掉舊容器

# 停掉舊容器后再部署新容器
docker stop <old_container_id>
docker run ...  # 啟動新容器

方案 2：用不同的 Stream 做測試

• 新容器監(jiān)聽不同的 Stream，例如 message:maf:tasks_test

• 測試時往 message:maf:tasks_test 發(fā)消息

• 需要改配置或環(huán)境變量，讓新容器使用 message:maf:tasks_test

方案 3：用不同的消費(fèi)者組

• 新容器使用不同的消費(fèi)者組名，例如 maf_analyze_group_v2

• 兩個組都會收到同一條消息，各自處理一次

• 適合做 A/B 或灰度，但會產(chǎn)生重復(fù)處理，需要業(yè)務(wù)上能接受

方案 4：只保留一個容器

• 部署新容器前先停掉舊容器

• 或使用滾動更新，保證同一時間只有一個版本在跑

六、Redis 常用配置

# 綁定地址
bind 0.0.0.0
# 端口
port 6379
# 密碼
requirepass your_password
# 最大內(nèi)存
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
# 啟動服務(wù)
sudo systemctl start redis
sudo systemctl enable redis
# 驗(yàn)證
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
# 遠(yuǎn)程連接
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 消費(fèi)
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. 合理設(shè)置 maxmemory 和淘汰策略，避免 OOM。
2. 生產(chǎn)環(huán)境開啟 requirepass 和訪問控制。
3. 使用連接池，減少連接開銷。
4. 對熱點(diǎn) key 做拆分或本地緩存，減輕壓力。
5. 使用 Pipeline 批量執(zhí)行命令，降低網(wǎng)絡(luò)往返。
6. Stream 場景下注意消費(fèi)者組和 ACK，避免消息堆積和重復(fù)消費(fèi)。

總結(jié)

Redis 適合做緩存、會話存儲和消息隊(duì)列。Stream 在需要可靠消費(fèi)、負(fù)載均衡和消息確認(rèn)的場景中，比簡單 List 更合適。結(jié)合具體業(yè)務(wù)（如 MAF 任務(wù)流），可以設(shè)計(jì)出清晰、可擴(kuò)展的異步處理架構(gòu)。

到此這篇關(guān)于Redis從基礎(chǔ)到Stream消息隊(duì)列實(shí)戰(zhàn)指南的文章就介紹到這了,更多相關(guān)redis stream 消息隊(duì)列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論