17-原理 6:小道消息 —— PubSub
课程
1
开篇:授人以鱼不若授人以渔 —— Redis 可以用来做什么?
学习时长: 5分21秒
2
基础:万丈高楼平地起 —— Redis 基础数据结构
上次学到
学习时长: 16分14秒
3
应用 1:千帆竞发 —— 分布式锁
学习时长: 7分47秒
4
应用 2:缓兵之计 —— 延时队列
学习时长: 8分9秒
5
应用 3:节衣缩食 —— 位图
学习时长: 8分52秒
6
应用 4:四两拨千斤 —— HyperLogLog
学习时长: 14分17秒
7
应用 5:层峦叠嶂 —— 布隆过滤器
学习时长: 17分54秒
8
应用 6:断尾求生 —— 简单限流
学习时长: 4分37秒
9
应用 7:一毛不拔 —— 漏斗限流
学习时长: 7分22秒
10
应用 8:近水楼台 —— GeoHash
学习时长: 7分52秒
11
应用 9:大海捞针 —— Scan
学习时长: 8分42秒
12
原理 1:鞭辟入里 —— 线程 IO 模型
学习时长: 4分1秒
13
原理 2:交头接耳 —— 通信协议
学习时长: 3分34秒
14
原理 3:未雨绸缪 —— 持久化
学习时长: 5分27秒
15
原理 4:雷厉风行 —— 管道
学习时长: 3分51秒
16
原理 5:同舟共济 —— 事务
学习时长: 6分36秒
17
原理 6:小道消息 —— PubSub
学习时长: 7分7秒
18
原理 7:开源节流 —— 小对象压缩
学习时长: 7分14秒
19
原理 8:有备无患 —— 主从同步
学习时长: 4分9秒
20
集群 1:李代桃僵 —— Sentinel
学习时长: 3分52秒
21
集群 2:分而治之 —— Codis
学习时长: 7分28秒
22
集群 3:众志成城 —— Cluster
学习时长: 8分38秒
23
拓展 1:耳听八方 —— Stream
学习时长: 13分40秒
24
拓展 2:无所不知 —— Info 指令
学习时长: 4分4秒
25
拓展 3:拾遗补漏 —— 再谈分布式锁
学习时长: 2分18秒
26
拓展 4:朝生暮死 —— 过期策略
学习时长: 2分21秒
27
拓展 5:优胜劣汰 —— LRU
学习时长: 4分34秒
28
拓展 6:平波缓进 —— 懒惰删除
学习时长: 2分13秒
29
拓展 7:妙手仁心 —— 优雅地使用 Jedis
学习时长: 6分35秒
30
拓展 8:居安思危 —— 保护 Redis
学习时长: 2分19秒
31
拓展 9:隔墙有耳 —— Redis 安全通信
学习时长: 6分34秒
32
拓展 10:法力无边 —— Redis Lua 脚本执行原理
学习时长: 9分24秒
33
拓展 11:短小精悍 —— 命令行工具的妙用
学习时长: 9分21秒
34
源码 1:丝分缕析 —— 探索「字符串」内部
学习时长: 5分20秒
35
源码 2:循序渐进 —— 探索「字典」内部
学习时长: 7分24秒
36
源码 3:挨肩迭背 —— 探索「压缩列表」内部
学习时长: 10分42秒
37
源码 4:风驰电掣 —— 探索「快速列表」内部
学习时长: 3分49秒
38
源码 5:凌波微步 —— 探索「跳跃列表」内部
学习时长: 9分57秒
39
源码 6:破旧立新 —— 探索「紧凑列表」内部
学习时长: 2分42秒
40
源码 7:金枝玉叶 —— 探索「基数树」内部
学习时长: 5分36秒
41
源码 8:精益求精 —— LFU vs LRU
学习时长: 8分4秒
42
源码 9:如履薄冰 —— 懒惰删除的巨大牺牲
学习时长: 9分53秒
43
源码 10:跋山涉水 —— 深入字典遍历
学习时长: 9分24秒
44
源码 11:见缝插针 —— 探索 HyperLogLog 内部
学习时长: 13分3秒
45
尾声:百尺竿头 —— 继续深造指南
学习时长: 2分32秒
juejin_logo copyCreated with Sketch.

原理 6:小道消息 —— PubSub

前面我们讲了 Redis 消息队列的使用方法,但是没有提到 Redis 消息队列的不足之处,那就是它不支持消息的多播机制

消息多播

消息多播允许生产者生产一次消息,中间件负责将消息复制到多个消息队列,每个消息队列由相应的消费组进行消费。它是分布式系统常用的一种解耦方式,用于将多个消费组的逻辑进行拆分。支持了消息多播,多个消费组的逻辑就可以放到不同的子系统中。

如果是普通的消息队列,就得将多个不同的消费组逻辑串接起来放在一个子系统中,进行连续消费。

PubSub

为了支持消息多播,Redis 不能再依赖于那 5 种基本数据类型了。它单独使用了一个模块来支持消息多播,这个模块的名字叫着 PubSub,也就是 PublisherSubscriber,发布者订阅者模型。我们使用 Python 语言来演示一下 PubSub 如何使用。

# -*- coding: utf-8 -*-
import time
import redis

client = redis.StrictRedis()
p = client.pubsub()
p.subscribe("codehole")
time.sleep(1)
print p.get_message()
client.publish("codehole", "java comes")
time.sleep(1)
print p.get_message()
client.publish("codehole", "python comes")
time.sleep(1)
print p.get_message()
print p.get_message()
{'pattern': None, 'type': 'subscribe', 'channel': 'codehole', 'data': 1L}
{'pattern': None, 'type': 'message', 'channel': 'codehole', 'data': 'java comes'}
{'pattern': None, 'type': 'message', 'channel': 'codehole', 'data': 'python comes'}
None

客户端发起订阅命令后,Redis 会立即给予一个反馈消息通知订阅成功。因为有网络传输延迟,在 subscribe 命令发出后,需要休眠一会,再通过 get\_message 才能拿到反馈消息。客户端接下来执行发布命令,发布了一条消息。同样因为网络延迟,在 publish 命令发出后,需要休眠一会,再通过 get\_message 才能拿到发布的消息。如果当前没有消息,get\_message 会返回空,告知当前没有消息,所以它不是阻塞的。

Redis PubSub 的生产者和消费者是不同的连接,也就是上面这个例子实际上使用了两个 Redis 的连接。这是必须的,因为 Redis 不允许连接在 subscribe 等待消息时还要进行其它的操作。

在生产环境中,我们很少将生产者和消费者放在同一个线程里。如果它们真要在同一个线程里,何必通过中间件来流转,直接使用函数调用就行。所以我们应该将生产者和消费者分离,接下来我们看看分离后的代码要怎么写。

消费者

# -*- coding: utf-8 -*-
import time
import redis

client = redis.StrictRedis()
p = client.pubsub()
p.subscribe("codehole")
while True:
    msg = p.get_message()
    if not msg:
        time.sleep(1)
        continue
    print msg

生产者

# -*- coding: utf-8 -*-
import redis

client = redis.StrictRedis()
client.publish("codehole", "python comes")
client.publish("codehole", "java comes")
client.publish("codehole", "golang comes")

必须先启动消费者,然后再执行生产者,消费者我们可以启动多个,pubsub 会保证它们收到的是相同的消息序列。

{'pattern': None, 'type': 'subscribe', 'channel': 'codehole', 'data': 1L}
{'pattern': None, 'type': 'message', 'channel': 'codehole', 'data': 'python comes'}
{'pattern': None, 'type': 'message', 'channel': 'codehole', 'data': 'java comes'}
{'pattern': None, 'type': 'message', 'channel': 'codehole', 'data': 'golang comes'}

我们从消费者的控制台窗口可以看到上面的输出,每个消费者窗口都是同样的输出。第一行是订阅成功消息,它很快就会输出,后面的三行会在生产者进程执行的时候立即输出。 上面的消费者是通过轮询 get_message 来收取消息的,如果收取不到就休眠 1s。这让我们想起了第 3 节的消息队列模型,我们使用 blpop 来代替休眠来提高消息处理的及时性。

PubSub 的消费者如果使用休眠的方式来轮询消息,也会遭遇消息处理不及时的问题。不过我们可以使用 listen 来阻塞监听消息来进行处理,这点同 blpop 原理是一样的。下面我们改造一下消费者

阻塞消费者

# -*- coding: utf-8 -*-
import time
import redis

client = redis.StrictRedis()
p = client.pubsub()
p.subscribe("codehole")
for msg in p.listen():
    print msg

代码简短了很多,不需要再休眠了,消息处理也及时了。

模式订阅

上面提到的订阅模式是基于名称订阅的,消费者订阅一个主题是必须明确指定主题的名称。如果我们想要订阅多个主题,那就 subscribe 多个名称。

> subscribe codehole.image codehole.text codehole.blog  # 同时订阅三个主题,会有三条订阅成功反馈信息
1) "subscribe"
2) "codehole.image"
3) (integer) 1
1) "subscribe"
2) "codehole.text"
3) (integer) 2
1) "subscribe"
2) "codehole.blog"
3) (integer) 3

这样生产者向这三个主题发布的消息,这个消费者都可以接收到。

> publish codehole.image https://www.google.com/dudo.png
(integer) 1
> publish codehole.text " 你好,欢迎加入码洞 "
(integer) 1
> publish codehole.blog '{"content": "hello, everyone", "title": "welcome"}'
(integer) 1

如果现在要增加一个主题codehole.group,客户端必须也跟着增加一个订阅指令才可以收到新开主题的消息推送。

为了简化订阅的繁琐,redis 提供了模式订阅功能Pattern Subscribe,这样就可以一次订阅多个主题,即使生产者新增加了同模式的主题,消费者也可以立即收到消息

> psubscribe codehole.*  # 用模式匹配一次订阅多个主题,主题以 codehole. 字符开头的消息都可以收到
1) "psubscribe"
2) "codehole.*"
3) (integer) 1

消息结构

前面的消费者消息输出时都是下面的这样一个字典形式

{'pattern': None, 'type': 'subscribe', 'channel': 'codehole', 'data': 1L}
{'pattern': None, 'type': 'message', 'channel': 'codehole', 'data': 'python comes'}
{'pattern': None, 'type': 'message', 'channel': 'codehole', 'data': 'java comes'}
{'pattern': None, 'type': 'message', 'channel': 'codehole', 'data': 'golang comes'}

那这几个字段是什么含义呢?

data 这个毫无疑问就是消息的内容,一个字符串。

channel 这个也很明显,它表示当前订阅的主题名称。

type 它表示消息的类型,如果是一个普通的消息,那么类型就是 message,如果是控制消息,比如订阅指令的反馈,它的类型就是 subscribe,如果是模式订阅的反馈,它的类型就是 psubscribe,还有取消订阅指令的反馈 unsubscribe 和 punsubscribe。

pattern 它表示当前消息是使用哪种模式订阅到的,如果是通过 subscribe 指令订阅的,那么这个字段就是空。

PubSub 缺点

PubSub 的生产者传递过来一个消息,Redis 会直接找到相应的消费者传递过去。如果一个消费者都没有,那么消息直接丢弃。如果开始有三个消费者,一个消费者突然挂掉了,生产者会继续发送消息,另外两个消费者可以持续收到消息。但是挂掉的消费者重新连上的时候,这断连期间生产者发送的消息,对于这个消费者来说就是彻底丢失了。

如果 Redis 停机重启,PubSub 的消息是不会持久化的,毕竟 Redis 宕机就相当于一个消费者都没有,所有的消息直接被丢弃。

正是因为 PubSub 有这些缺点,它几乎找不到合适的应用场景。所以 Redis 的作者单独开启了一个项目 Disque 专门用来做多播消息队列。该项目目前没有成熟,一直长期处于 Beta 版本,但是相应的客户端 sdk 已经非常丰富了,就待 Redis 作者临门一脚发布一个 Release 版本。关于 Disque 的更多细节,本小册不会多做详细介绍,感兴趣的同学可以去阅读相关文档。

补充

近期 Redis5.0 新增了 Stream 数据结构,这个功能给 Redis 带来了持久化消息队列,从此 PubSub 可以消失了,Disqueue 估计也永远发不出它的 Release 版本了。具体内容请读者阅读 Stream 章节内容(第 23 节)。

留言
Ctrl + Enter
全部评论(18)
柏油的头像
删除
JAVA后端
梳理了客户端 -> 服务端的处理流程,大家感兴趣可以看看blog.csdn.net
点赞
回复
BinK_1783的头像
删除
喜欢是没有生产日期的,也没有保质期
点赞
回复
加勒比海参的头像
删除
Java
redisson的分布式锁还在用这个
点赞
回复
无花9999的头像
删除
机密
全文没写mq消息监听 差评
点赞
回复
zyfsuzy的头像
删除
研发工程师 @ 百度
消息队列还是得用mq
点赞
回复
keepstyler的头像
删除
全栈
有多个消费者时,怎么实现消息只允许被其中一个消费者消费
点赞
2
删除
需要新开一个生产者吧,让消费者订阅新的生产者。
点赞
回复
删除
这是要想进行权限控制么?
点赞
回复
pengmingfa521的头像
删除
学习了
点赞
回复
钱同志的头像
删除
Redis的初衷是解决应用的性能问题,虽然,也有一些针对特定场景的功能,不过有些比较鸡肋,小需求能满足,完善的功能就不如更专业的其他中间件了!
3
回复
随心而至的头像
删除
如果是模式订阅的消息type是pmessage
点赞
回复
dre52826的头像
删除
消息还是要专门的消息组件来做比较合适,有时间尝试一下Stream
点赞
回复
M。同学的头像
删除
订阅者集群模式,有没有一种方案能够保证集群中所有的节点都能确认无误的收到信息
3
1
删除
redis主要是缓存....mq不是他的强项
点赞
回复
LuckyStar23057的头像
删除
消费者会出现超时断开的问题吗?
点赞
1
删除
客户端会做处理,一般连接不会断开。客户端会做一些检测机制,即使断开了也会重新连接(重新订阅)
点赞
回复
Zzde的头像
删除
可以讲讲键空间通知的用途吗?
1
回复
LeGrandmechantRenard的头像
删除
想吃软饭 @ 某支付公司
可惜了 还想做个im呢。
点赞
回复