Redis实战（5）——Redis实现消息队列
消息队列，顾名思义，就是一个存放消息的队列。最简单的消息队列包含3个角色
- 生产者：将消息存入队列中
- 队列：存放和管理消息
- 消费者： 将消息从队列中取出来并做业务处理
R
e
d
i
s
提供了三种实现消息队列的方式，基于
L
i
s
t
结构、
P
u
b
S
u
b
、
S
t
r
e
a
m
结构
\textcolor{red}{Redis 提供了三种实现消息队列的方式，基于List结构、PubSub、Stream结构}
Redis提供了三种实现消息队列的方式，基于List结构、PubSub、Stream结构
1 基于List 结构实现消息队列
Redis的List是一个双向列表。可以从两端存入数据或者取出数据。
- LPUSH/RPUSH key element 【elements】
- BLPOP/RPOP key timeout
利用list 结构实现的消息队列主要是依据阻塞取指令 BLPOP/RPOP 来模拟消费者监听队列，直到队列中有消失时获得该数据
优点： 实现简单，且可以持久化
缺点: 只能有一个消费者来消费数据，且只能消费一次,无法避免消息的丢失
2 基于PubSub（发布/订阅）
PubSub 是一个基于点对点的消息模型，消费者可以订阅一个或者多个chanel，当生产者向队列发送了消息时，消费者只要订阅了频道就可以收到并处理消息
- PUBLISH channel message 将信息 message 发送到指定的频道 channel
- SUBSCRIBE channel [channel …] 订阅一个或多个频道
- PSUBSCRIBE pattren 订阅与通配符匹配的chanel
在使用PSUBSCRIBE pattren 时，支持多种通配符
1
\textcolor{blue}{1}
1 ？：匹配一个字符
2
\textcolor{blue}{2}
2 * ：匹配零个字符或多个字符
3
\textcolor{blue}{3}
3 [] ：选择匹配，匹配[]中定义的字符 如hell[ae]o 可以匹配 hello 和 hellao
使用PubSub 实现的消费队列时，支持
多生产、多消费
\textcolor{red}{多生产、多消费}
多生产、多消费的模式，不过PubSub不支持数据的持久化，相较于List，它本身就不是一个数据结构无法利用Redis持久化数据。并且无法避免消息的丢失，如生产者向无人订阅的频道发消息时，数据会丢失。另外还会出现由于消费者的缓存空间有效，超时缓存上限时，将会出现消息的丢失。由于这些缺点，redis的PUBSUB模式，无法满足对可靠性要求较高的服务。
3 基于Stream 数据结构
Stream 是redis5.0 及之后针对消息队列场景设计的
数据结构
\textcolor{red}{数据结构}
数据结构，因此数据的安全性得到了保障，因为可以持久化。相较于List 数据结构实现的消息队列的方式，有更多针对消息队列的单独命令，可以实现一个功能更加完善的消息队列
发送消息
- XADD
k
e
y
\textcolor{red}{key}
key
[
N
O
M
K
S
T
R
E
A
M
]
\textcolor{blue}{[NOMKSTREAM] }
[NOMKSTREAM]
[
[
=
∣
]
t
h
r
e
s
h
o
l
d
[
L
I
M
I
T
c
o
u
n
t
]
]
\textcolor{green}{[ [= | ~] threshold [LIMIT count]]}
[[=∣ ]threshold[LIMITcount]]
\textcolor{orange}{}
F
i
e
l
d
v
a
l
u
e
[
F
i
e
l
d
v
a
l
u
e
.
.
.
]
\textcolor{purple}{Field value [Field value …]}
Fieldvalue[Fieldvalue…]
参数说明
K
e
y
\textcolor{red}{Key}
Key : 存储消息的队列的名字
[
N
O
M
K
S
T
R
E
A
M
]
\textcolor{blue}{[NOMKSTREAM] }
[NOMKSTREAM] ：可选参数，是否在队列不存在时，创建队列。默认是创建的
[
[
=
∣
]
t
h
r
e
s
h
o
l
d
[
L
I
M
I
T
c
o
u
n
t
]
]
\textcolor{green}{[ [= | ~] threshold [LIMIT count]]}
[[=∣ ]threshold[LIMITcount]] :可选参数，设置消息队列的最大消息数，默认是设上限的
\textcolor{orange}{}
：消息的唯一id，* 表示有redis自动生成，格式是时间戳_递增值 如 1526919030474-0。Id值也可以自定义。
F
i
e
l
d
v
a
l
u
e
[
F
i
e
l
d
v
a
l
u
e
.
.
.
]
\textcolor{purple}{Field value [Field value …]}
Fieldvalue[Fieldvalue…] 消息体
读取消息
- XREAD
[
C
O
U
N
T
c
o
u
n
t
]
\textcolor{red}{[COUNT count] }
[COUNTcount]
[
B
L
O
C
K
m
i
l
l
i
s
e
c
o
n
d
s
]
\textcolor{blue}{[BLOCK milliseconds]}
[BLOCKmilliseconds]
S
T
R
E
A
M
S
k
e
y
[
k
e
y
.
.
.
]
\textcolor{green}{STREAMS key [key …] }
STREAMSkey[key…]
i
d
[
i
d
.
.
.
]
\textcolor{orange}{id [id …]}
id[id…]
参数说明
[
C
O
U
N
T
c
o
u
n
t
]
\textcolor{red}{[COUNT count] }
[COUNTcount] 可选参数， 指定读取消息的条数
[
B
L
O
C
K
m
i
l
l
i
s
e
c
o
n
d
s
]
\textcolor{blue}{[BLOCK milliseconds]}
[BLOCKmilliseconds] 当没有消息时，读取队列消息的阻塞时长，当设置为0时，永久等待，直到读取到队列中消息
S
T
R
E
A
M
S
k
e
y
[
k
e
y
.
.
.
]
\textcolor{green}{STREAMS key [key …] }
STREAMSkey[key…] 需要读取的队列的key名字，可以从多个队列中读取数据
i
d
[
i
d
.
.
.
]
\textcolor{orange}{id [id …]}
id[id…] 读取消息的起始Id 。有两个特殊的id，0 表示从第一个消息读起，$ 表示读取最新的一条消息
在读取消息时，可以通过while(true) 循环 调用xread block 0 streams key $ 去永久的监听队列去获得消息。不过这种模式下会出现一个问题，在获得消息并处理消息这个时间间隙中，可能生产者又往队列中增加了好几条消息，由于Id 为$ 只会读取最新的一条消息，那么可能会出现消息的漏读。这里可以采用基于消费者组去读取消息
3.1 基于消费者组去消费消息
可以将多个消费者划分到一个组中，其中每个组消费消息时都会维护一个最后消费消息的标识
L
a
s
t
d
e
l
i
v
e
r
e
d
i
d
\textcolor{red}{Last delivered id}
Lastdeliveredid，当宕机重启后，直接从该标识id之后的消息消费。意味者不会重复消费消息。
在消费者组中还维护了一个 Pending_ids集合，该集合中存放了未确认【ACK】消费数据的消息Id，
机器出现宕机后重启，可继续确认未处理的消息。可以通过
X
A
C
K
\textcolor{red}{XACK}
XACK来确认客户端确认已经消费了消息，之后从Pending_ids集合中移除。
基于消费者组消费消息时，最大程度的保证了消息的安全消费、不重复消费。
创建消费者组
XGROUP
C
R
E
A
T
E
\textcolor{red}{CREATE}
CREATE
K
E
Y
\textcolor{green}{KEY }
KEY
G
R
O
U
P
N
A
M
E
\textcolor{blue}{GROUPNAME }
GROUPNAME
I
D
\textcolor{orange}{ID}
ID
[
M
K
S
T
R
E
A
M
]
\textcolor{purple}{ [MKSTREAM]}
[MKSTREAM]
C
R
E
A
T
E
\textcolor{red}{CREATE}
CREATE ：创建组
K
E
Y
\textcolor{green}{KEY }
KEY ：基于哪个队列去创建组
G
R
O
U
P
N
A
M
E
\textcolor{blue}{GROUPNAME }
GROUPNAME ：创建的消费者组名称
I
D
\textcolor{orange}{ID}
ID 消息的标识id。0 从头消费 $ 消费最新的消息
[
M
K
S
T
R
E
A
M
]
\textcolor{purple}{ [MKSTREAM]}
[MKSTREAM] ： 可选参数，当队列不存在时，是否创建队列，默认是创建
从消费者组中消费消息
XGROUPREAD GROUP
g
r
o
u
p
\textcolor{red}{group }
group
c
o
n
s
u
m
e
r
\textcolor{green}{consumer }
consumer
[
C
O
U
N
T
c
o
u
n
t
]
\textcolor{blue}{[COUNT count] }
[COUNTcount]
[
B
l
o
c
k
m
i
l
l
i
s
e
c
o
n
d
s
]
\textcolor{orange}{[Block milliseconds] }
[Blockmilliseconds]
[
N
O
A
C
K
]
\textcolor{purple}{ [NOACK] }
[NOACK]
S
T
R
E
A
M
S
K
E
Y
[
k
e
y
.
.
.
]
\textcolor{red}{STREAMS KEY [key …]}
STREAMSKEY[key…]
I
D
[
I
D
.
.
.
.
.
]
\textcolor{green}{ ID [ID…..] }
ID[ID…..]
g
r
o
u
p
\textcolor{red}{group }
group : 组的名字，定义从哪个消费者组消费消息
c
o
n
s
u
m
e
r
\textcolor{green}{consumer }
consumer ：消费者名字，如果不存在，自动创建
[
C
O
U
N
T
c
o
u
n
t
]
\textcolor{blue}{[COUNT count] }
[COUNTcount] ：消费数量
[
B
l
o
c
k
m
i
l
l
i
s
e
c
o
n
d
s
]
\textcolor{orange}{[Block milliseconds] }
[Blockmilliseconds] ：可选参数，阻塞时长【单位ms】，不设置时为非阻塞消费。
[
N
O
A
C
K
]
\textcolor{purple}{ [NOACK] }
[NOACK] ：可选参数，是否自动确认。true时消息不会进入pending_ids[] 集合中，可能会有未消费的消息。所以为了安全性，无需设置。
S
T
R
E
A
M
S
K
E
Y
[
k
e
y
.
.
.
]
\textcolor{red}{STREAMS KEY [key …]}
STREAMSKEY[key…] ： 监听的队列的名字
I
D
[
I
D
.
.
.
.
.
]
\textcolor{green}{ ID [ID…..] }
ID[ID…..] ：获得消息的起始ID 。
设置成 “>” ：从下一个未消费的消息开始消费。
设置成其他：均是从pending-list中获得已消费但是未确认的消息，如0 ，从pending-list中第一个消息开始。
根据实际情况可设置不同的ID 去消费消息。正常读取设置> 异常读取未确认的消息
确认消息
XACK
k
e
y
\textcolor{red}{key }
key
g
r
o
u
p
\textcolor{green}{group }
group
I
D
[
I
D
.
.
.
]
\textcolor{blue}{ ID [ID…] }
ID[ID…]
k
e
y
\textcolor{red}{key }
key ：队列名称
g
r
o
u
p
\textcolor{green}{group }
group ：组名称
I
D
[
I
D
.
.
.
]
\textcolor{blue}{ ID [ID…] }
ID[ID…] ：待确认的消息Id
3.2 数据测试
(1) 向order队列中添加4条消息
xadd order * voucherId 9 userId 150 orderId 79297921056506055 xadd order * voucherId 9 userId 129 orderId 79297921056506083 xadd order * voucherId 9 userId 111 orderId 79297921056506108 xadd order * voucherId 9 userId 111 orderId 79297921056506101
(2) 向order队列创建消费者组group_1
## 消费者组从头开始消费数据 XGROUP CREATE order group_1 0
(3) 从消费者组中消费消息
## 消费最新的未消费的消息，采用阻塞式获取，最长等待2000ms XREADGROUP GROUP group_1 consumer_1 COUNT 1 BLOCK 2000 STRAEAMS order >
第五次消费时，阻塞等待后返回空。队列中的消息全部消费，此时都处于为确认状态，全部存入了penging-list中。
此时需要手动确认这些消息确实已经被成功的消费了，需要手动确认将其从pending-list 集合中移除
(4) 手动确认已经消费的消息
XACK order group_1 1691146911471-0 1691148054821-0 1691148657217-0 1691202770386-0
本文来自网络，不代表协通编程立场，如若转载，请注明出处：https://net2asp.com/fe8e108808.html
相关推荐
-
Python之Pandas的常用技能【写入数据】
1、背景： 最近在工作中遇到越来越多的的使用pandas或者python来处里写入操作，尤其是对excel文件或者csv文件的操作更是常见，这里将写入操作总结如下，方便记忆，也分享…
-
【数据结构】直接插入排序 & 希尔排序（一）
目录 一，排序的概念 二，直接插入排序 1，基本思想 2，基本思路 3，思路实现 三，希尔排序 1，希尔排序的特性总结： 2，思路实现： 一，排序的概念 排序：所谓排序，就是使一串…
-
Python–从PDF中提取文本的方法总结
目录 前言 一、pdfplumber 二、pdfminer 三、fitz / pymupdf 四、性能对比 前言 这段时间做了好几个关于年报的需求，其中无一例外需要从年报PDF中提…
-
【大虾送书第三期】《Python高并发与高性能编程: 原理与实践》
目录 ✨写在前面 ✨主要内容 ✨本书特色 ✨关于作者 🦐博客主页：大虾好吃吗的博客 🦐专栏地址：免费送书活动专栏地址 写在前面 …
-
Python-Django的“日志功能-日志模块(logging模块)-日志输出”的功能详解
提前说明：如果是简单的日志输出，看我的另一篇博客就够，链接如下： 提前说明：如果是简单的日志输出，看我的另一篇博客就够，链接如下： https://blog.csdn.net/we…
-
YOLOV5配置至运行（本人亲装遇见常见的错误及解决方案）
现在是2023年3月26日，由于最近有个比赛需要用到yolov5但是本人从前从未接触此方面知识，想着先给他跑起来再说。于是本人亲自装在自己电脑上，准备跑一下，并且和大家一样踩了很多…
-
Python的离线安装
原文链接 在没有外网的情况下，安装Python环境只能采用离线方式。 Windows离线安装Python Python离线安装包的下载地址：https://www.python.o…
-
【 Python ModuleNotFoundError: No module named ‘xxx‘可能的解决方案大全】
Python ModuleNotFoundError: No module named ‘xxx‘可能的解决方案大全 本文主要介绍了Python ModuleNotFoundErr…
-
C++（Qt）与Python混合编程
文章开头，请大家注意一点： 开始搭配环境之前，请大家完成对自己电脑位数的查看【32位/64位】 本篇博客适用于windows64位版本安装。当然了，32位的友友也可以对照着配置！ …
-
偏最小二乘(PLS)原理分析&Python实现
目录 1 偏最小二乘的意义 2 PLS实现步骤 3 弄懂PLS要回答的问题 4 PLS的原理分析 4.1 自变量和因变量的主成分求解原理 4.1.1 确定目标函数 …
