一、Redis 簡(jiǎn)介

Redis（Remote Dictionary Server）是一個(gè)基于內(nèi)存的鍵值存儲(chǔ)系統(tǒng)，支持多種數(shù)據(jù)結(jié)構(gòu)，常被用作緩存、消息隊(duì)列和會(huì)話存儲(chǔ)。它提供高性能讀寫(xiě)，并支持持久化、主從復(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è)置過(guò)期時(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      # 范圍查詢(xún)

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 的對(duì)比

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

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

5.1 架構(gòu)示意

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

5.2 MAF 營(yíng)銷(xiāo)分析任務(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)聽(tīng) message:maf:tasks，消費(fèi)并分析
# 3. 引擎完成后寫(xiě)入結(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)者組帶來(lái)的好處

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

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

• 支持 ACK，失敗可重試

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

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

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

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

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

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

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

因此：

• 有時(shí)是A容器處理

• 有時(shí)是B容器處理

• 測(cè)試時(shí)無(wú)法穩(wěn)定地“只讓某個(gè)容器”處理請(qǐng)求

可能帶來(lái)的問(wèn)題：

如何保證只使用新代碼：

方案 1：停掉舊容器

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

方案 2：用不同的 Stream 做測(cè)試

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

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

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

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

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

• 兩個(gè)組都會(huì)收到同一條消息，各自處理一次

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

方案 4：只保留一個(gè)容器

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

• 或使用滾動(dòng)更新，保證同一時(shí)間只有一個(gè)版本在跑

六、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
# 啟動(dòng)服務(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)境開(kāi)啟 requirepass 和訪問(wèn)控制。
3. 使用連接池，減少連接開(kāi)銷(xiāo)。
4. 對(duì)熱點(diǎn) key 做拆分或本地緩存，減輕壓力。
5. 使用 Pipeline 批量執(zhí)行命令，降低網(wǎng)絡(luò)往返。
6. Stream 場(chǎng)景下注意消費(fèi)者組和 ACK，避免消息堆積和重復(fù)消費(fèi)。

總結(jié)

Redis 適合做緩存、會(huì)話存儲(chǔ)和消息隊(duì)列。Stream 在需要可靠消費(fèi)、負(fù)載均衡和消息確認(rèn)的場(chǎng)景中，比簡(jiǎ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)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論