Redis入门第四步：Redis发布与订阅

欢迎继续跟随《Redis新手指南：从入门到精通》专栏的步伐！在本文中，我们将深入探讨Redis的发布与订阅（Pub/Sub）模式。这是一种强大的消息传递机制，适用于各种实时通信场景，如聊天应用、实时通知和事件驱动架构。通过学习本篇文章，你将能够掌握如何使用Redis的Pub/Sub功能来构建高效的消息传递系统。

在本文中，你将会学到：

为什么需要发布与订阅模式: 通过业务实际案例解释为什么需要Pub/Sub模式
什么是发布与订阅模式：理解Pub/Sub的基本概念及其在分布式系统中的重要性。
Redis的Pub/Sub命令：详细介绍如何使用PUBLISH, SUBSCRIBE, UNSUBSCRIBE, 和 PSUBSCRIBE等命令。
频道与模式：了解频道（channels）和模式（patterns）的区别及用法。
实战案例：通过一个简单的聊天应用示例，展示如何在实际项目中使用Redis的Pub/Sub功能。
最佳实践：分享一些使用Pub/Sub时的最佳实践，包括性能优化和故障处理。

无论你是初学者还是有经验的开发者，本文都将为你提供清晰易懂的指导，帮助你在项目中有效利用Redis的发布与订阅功能。让我们一起探索这一强大的消息传递机制吧！

也许有的小伙伴对这个功能比较陌生，不太清楚这个功能是干什么的. 不妨我们先举个案例

假设我们有这么一个业务场景，在网站下单支付以后，需要通知库存服务进行发货处理。

上面业务实现不难，我们只要让库存服务提供给相关的给口，下单支付之后只要调用库存服务即可。

后面如果又有新的业务，比如说积分服务，他需要获取下单支付的结果，然后增加用户的积分。

这个实现也不难，让积分服务同样提供一个接口，下单支付之后只要调用库存服务即可。

如果就两个业务需要获取下单支付的结果，那也还好，程序改造也快。可是随着业务不断的发展，越来越多的新业务说是要下单支付的结果。

这时我们会发现上面这样的系统架构存在很多问题：

第一，下单支付业务与其他业务重度耦合，每当有个新业务需要支付结果，就需要改动下单支付的业务。

第二，如果调用业务过多，会导致下单支付接口响应时间变长。另外，如果有任一下游接口响应变慢，就会同步导致下单支付接口响应也变长。

第三，如果任一下游接口失败，可能导致数据不一致的情况。比如说下图，先调用 A，成功之后再调用 B，最后再调用 C。

如果在调用 B 接口的发生异常，此时可能就导致下单支付接口返回失败，但是此时 A 接口其实已经调用成功，这就代表它内部已经处理下单支付成功的结果。

这样就会导致 A，B，C 三个下游接口，A 获取成功获取支付结果，但是 B，C 没有拿到，导致三者系统数据不一致的情况。

其实我们仔细想一下，对于下单支付业务来讲，它其实不需要关心下游调用结果，只要有某种机制通知能通知到他们就可以了。

讲到这里，这就需要引入今天需要介绍发布订阅机制

1 什么是发布订阅🍀

Redis 发布订阅(pub/sub) 是一种消息通信模式: 发送者(pub)发送消息，订阅者(sub)接收消息。Redis 的 subscribe 命令可以让客户端订阅任意数量的频道，每当有新信息发送到被订阅的频道时，信息就会被发送给所有订阅指定频道的客户端

下图展示了频道channel ，以及订阅这个频道的三个客户端 —— client2 、 client5 和 client1 之间的关系:

当有新消息通过 publish 命令发送给频道 channel 时，这个消息就会被发送给订阅它的三个客户端：

如上图所示，消息发布者只需要想指定的频道发布消息，订阅该频道的每个客户端都可以接受到到这个消息。

使用 Redis 发布订阅这种机制，对于上面业务，下单支付业务只需要向「支付结果」这个频道发送消息，其他下游业务订阅「支付结果」这个频道，就能收相应消息，然后做出业务处理即可

这样就可以解耦系统上下游之间调用关系

2 为什么要用发布订阅🍀

针对消息订阅发布功能，市面上很多大厂使用的是kafka、RabbitMQ、ActiveMQ, RocketMQ等这几种，redis的订阅发布功能跟这三者相比，相对轻量，针对数据准确和安全性要求没有那么高可以直接使用，适用于小公司

1.双端队列

redis 的List数据类型结构提供了 blpop 、brpop 命令结合 rpush、lpush 命令可以实现消息队列机制，基于双端链表实现的发布与订阅功能

这种方式存在两个局限性：

不能支持一对多的消息分发。
如果生产者生成的速度远远大于消费者消费的速度，易堆积大量未消费的消息

◇ 双端队列图解如下：

✦ 解析：双端队列模式只能有一个或多个消费者轮着去消费，却不能将消息同时发给其他消费者

2. 发布/订阅模式

发布/订阅模式图解如下：

✦ 解析：redis订阅发布模式，生产者生产完消息通过频道分发消息，给订阅了该频道的所有消费

3 发布/订阅如何使用🍀

Redis有两种发布/订阅模式

1.基于频道(Channel)的发布/订阅

允许用户订阅一个或多个频道，以接收所有发送到这些频道的消息。这种模型与其它消息队列不同，因为它具有消息的持久性，即使消费者在消息发送时不在线，当它们上线时，它们仍然可以接收到所有的消息

命令	命令描述
subscribe channel [cha...]	订阅给定的一个或多个频道
unsubscribe channel [cha...]	退订给定的频道.说明:若没有指定channel,则默认退订所有频道
publish channel message	将消息发送给指定频道的channel，返回结果:接收到信息的订阅者数量，无订阅者返回0
publish channels [argument [arg]....]	查看订阅与发布系统的状态.说明:返回活跃频道列表(即至少有一个订阅者的频道,订阅模式的客户端除外)

"发布/订阅" 包含2种角色：发布者和订阅者。发布者可以向指定的频道(channel)发送消息；订阅者可以订阅一个或者多个频道(channel)，所有订阅此频道的订阅者都会收到此消息

订阅者订阅频道 subscribe channel [channel ...]

--------------------------客户端1（订阅者） ：订阅频道 ---------------------
 
# 订阅 “meihuashisan” 和 “csdn” 频道（如果不存在则会创建频道）
127.0.0.1:6379> subscribe meihuashisan csdn 
Reading messages... (press Ctrl-C to quit)
 
1) "subscribe"    -- 返回值类型：表示订阅成功！
2) "meihuashisan" -- 订阅频道的名称
3) (integer) 1    -- 当前客户端已订阅频道的数量
 
1) "subscribe"
2) "csdn"
3) (integer) 2
 
#注意：订阅后，该客户端会一直监听消息，如果发送者有消息发给频道，这里会立刻接收到消息

发布者发布消息 publish channel message

-----------------------客户端2（发布者）：发布消息给频道 -------------------

# 给“meihuashisan”这个频道 发送一条消息：“I am meihuashisan”
127.0.0.1:6379> publish meihuashisan "I am meihuashisan"
(integer) 1  # 接收到信息的订阅者数量，无订阅者返回0

客户端2(发布者)发布消息给频道后，此时我们再来观察客户端1(订阅者)的客户端窗口变化

--------------------------客户端1（订阅者） ：订阅频道 -----------------
127.0.0.1:6379> subscribe meihuashisan csdn 
Reading messages... (press Ctrl-C to quit)
 
1) "subscribe"    -- 返回值类型：表示订阅成功！
2) "meihuashisan" -- 订阅频道的名称
3) (integer) 1    -- 当前客户端已订阅频道的数量
 
1) "subscribe"
2) "csdn"
3) (integer) 2
 
--------------------变化如下：（实时接收到了该频道的发布者的消息）------------
1) "message"           -- 返回值类型：消息
2) "meihuashisan"      -- 来源（从哪个频道发过来的）
3) "I am meihuashisan" -- 消息内容

注意：如果是先发布消息，再订阅频道，不会收到订阅之前就发布到该频道的消息！

注意：进入订阅状态的客户端，不能使用除了subscribe、unsubscribe、psubscribe 和 punsubscribe 这四个属于"发布/订阅"之外的命令，否则会报错！

这里的客户端指的是 jedis、lettuce的客户端，redis-cli是无法退出订阅状态的

底层实现逻辑

底层通过字典实现。pubsub_channels 是一个字典类型，保存订阅频道的信息：字典的key为订阅的频道，字典的value是一个链表，链表中保存了所有订阅该频道的客户端

struct redisServer { 
  /* General */ 
  pid_t pid; 
  
  //省略百十行 
 
  // 将频道映射到已订阅客户端的列表(就是保存客户端和订阅的频道信息)
  dict *pubsub_channels; /* Map channels to list of subscribed clients */ 
}

数据结构

比如说，在下图展示的这个 pubsub_channels 示例中， client2 、 client5 和 client1 就订阅了 channel1 ，而其他频道也分别被别的客户端所订阅：

当客户端调用 SUBSCRIBE 命令时，程序就将客户端和要订阅的频道在 pubsub_channels 字典中关联起来。

举个例子，如果客户端 client10086 执行命令 SUBSCRIBE channel1 channel2 channel3 ，那么前面展示的 pubsub_channels 将变成下面这个样子：

发布

当调用 PUBLISH channel message 命令，程序首先根据 channel 定位到字典的键，然后将信息发送给字典值链表中的所有客户端。

比如说，对于以下这个 pubsub_channels 实例，如果某个客户端执行命令 PUBLISH channel1 "hello moto" ，那么 client2 、 client5 和 client1 三个客户端都将接收到 "hello moto" 信息：

退订

使用 UNSUBSCRIBE 命令可以退订指定的频道，这个命令执行的是订阅的反操作：它从 pubsub_channels 字典的给定频道（键）中，删除关于当前客户端的信息，这样被退订频道的信息就不会再发送给这个客户端。

2.基于模式(pattern)的发布/订阅

命令	命令描述
psubscribe pattern1 [pattern....]	订阅一个或多个符合给定模式的频道.说明:每个模式以为匹配符;例如cn匹配所有以cn开头的频道
punsubscribe pattern1 [pattern....]	退订所有给定模式的频道.说明:pattern未指定，则订阅的所有模式都会被退订，否则只退订指定的订阅模式

如果有某个/某些模式和该频道匹配，所有订阅这个/这些频道的客户端也同样会收到信息

图解

下图展示了一个带有频道和模式的例子，其中 com.ahead.* 频道匹配了 com.ahead.juc 频道和 com.ahead.thread 频道，并且有不同的客户端分别订阅它们三个，如下图：

当有信息发送到com.ahead.thread 频道时，信息除了发送给 client 4 和 client 5 之外，还会发送给订阅 com.ahead.* 频道模式的 client x 和 client y

✦ 解析：反之也是，如果当有消息发送给 com.ahead.juc 频道，消息发送给订阅了 juc 频道的客户端之外，还会发送给订阅了 com.ahead.* 频道的客户端： client x 、client y

通配符中?表示1个占位符，表示任意个占位符(包括0)，?表示1个以上占位符。

订阅者订阅频道 psubscribe pattern [pattern ...]

--------------------------客户端1（订阅者） ：订阅频道 --------------------
 
#  1. ------------订阅 “a?” "com.*" 2种模式频道--------------
127.0.0.1:6379> psubscribe a? com.*
# 进入订阅状态后处于阻塞，可以按Ctrl+C键退出订阅状态
Reading messages... (press Ctrl-C to quit) 
 
---------------订阅成功-------------------
 
1) "psubscribe"  -- 返回值的类型：显示订阅成功
2) "a?"          -- 订阅的模式
3) (integer) 1   -- 目前已订阅的模式的数量
 
1) "psubscribe"
2) "com.*"
3) (integer) 2
 
---------------接收消息 （已订阅 “a?” "com.*" 两种模式！）-----------------
 
# ---- 发布者第1条命令：publish ahead "hello"
结果：没有接收到消息，匹配失败，不满足 “a?” ，“？”表示一个占位符， a后面的head有4个占位符
 
# ---- 发布者第2条命令：  publish aa "hello" （满足 “a?”）
1) "pmessage" -- 返回值的类型：信息
2) "a?"       -- 信息匹配的模式：a?
3) "aa"       -- 信息本身的目标频道：aa
4) "hello"    -- 信息的内容："hello"
 
# ---- 发布者第3条命令：publish com.juc "hello2"（满足 “com.*”, *表示任意个占位符）
1) "pmessage" -- 返回值的类型：信息
2) "com.*"    -- 匹配模式：com.*
3) "com.juc"  -- 实际频道：com.juc
4) "hello2"   -- 信息："hello2"
 
---- 发布者第4条命令：publish com. "hello3"（满足 “com.*”, *表示任意个占位符）
1) "pmessage" -- 返回值的类型：信息
2) "com.*"    -- 匹配模式：com.*
3) "com."     -- 实际频道：com.
4) "hello3"   -- 信息："hello3"

发布者发布消息 publish channel message

------------------------客户端2（发布者）：发布消息给频道 ------------------
 
注意：订阅者已订阅 “a?” "com.*" 两种模式！
 
# 1. ahead 不符合“a?”模式，?表示1个占位符
127.0.0.1:6379> publish ahead "hello"  
(integer) 0    -- 匹配失败，0:无订阅者
 
# 2. aa 符合“a?”模式，?表示1个占位符
127.0.0.1:6379> publish aa "hello"      
(integer) 1
 
# 3. 符合“com.*”模式，*表示任意个占位符
127.0.0.1:6379> publish com.juc "hello2" 
(integer) 1
 
# 4. 符合“com.*”模式，*表示任意个占位符
127.0.0.1:6379> publish com. "hello3" 
(integer) 1

底层是pubsubPattern节点的链表。

数据结构 redisServer.pubsub_patterns 属性是一个链表，链表中保存着所有和模式相关的信息：

struct redisServer {
    // ...
    list *pubsub_patterns;
    // ...
};

链表中的每个节点都包含一个 redis.h/pubsubPattern 结构：

typedef struct pubsubPattern {
    redisClient *client;
    robj *pattern;
} pubsubPattern;

client 属性保存着订阅模式的客户端，而 pattern 属性则保存着被订阅的模式。

每当调用 PSUBSCRIBE 命令订阅一个模式时，程序就创建一个包含客户端信息和被订阅模式的 pubsubPattern 结构，并将该结构添加到 redisServer.pubsub_patterns 链表中。

作为例子，下图展示了一个包含两个模式的 pubsub_patterns 链表，其中 client123 和 client256 都正在订阅 tweet.shop.* 模式：

如果这时客户端 client10086 执行 PSUBSCRIBE broadcast.list.* ，那么 pubsub_patterns 链表将被更新成这样：

通过遍历整个 pubsub_patterns 链表，程序可以检查所有正在被订阅的模式，以及订阅这些模式的客户端。

发布

发送信息到模式的工作也是由 PUBLISH 命令进行的, 显然就是匹配模式获得Channels，然后再把消息发给客户端。

退订

使用 PUNSUBSCRIBE 命令可以退订指定的模式，这个命令执行的是订阅模式的反操作：程序会删除 redisServer.pubsub_patterns 链表中，所有和被退订模式相关联的 pubsubPattern 结构，这样客户端就不会再收到和模式相匹配的频道发来的信息

4 Redis 发布订阅实际应用🍀

Redis Sentinel 节点发现

「Redis Sentinel」 是 Redis 一套高可用方案，可以在主节点故障的时候，自动将从节点提升为主节点，从而转移故障。

今天这里我们不详细解释 「Redis Sentinel」 详细原理，主要来看下 「Redis Sentinel」 如何使用发布订阅机制。

「Redis Sentinel」 节点主要使用发布订阅机制，实现新节点的发现，以及交换主节点的之间的状态。

如下所示，每一个 「Sentinel」 节点将会定时向 _sentinel_:hello 频道发送消息，并且每个 「Sentinel」 都会订阅这个节点。

这样一旦有节点往这个频道发送消息，其他节点就可以立刻收到消息。

这样一旦有的新节点加入，它往这个频道发送消息，其他节点收到之后，判断本地列表并没有这个节点，于是就可以当做新的节点加入本地节点列表。

除此之外，每次往这个频道发送消息内容可以包含节点的状态信息，这样可以作为后面 「Sentinel」 领导者选举的依据。

以上都是对于 Redis 服务端来讲，对于客户端来讲，我们也可以用到发布订阅机制。

当 「Redis Sentinel」 进行主节点故障转移，这个过程各个阶段会通过发布订阅对外提供。

对于我们客户端来讲，比较关心切换之后的主节点，这样我们及时切换主节点的连接（旧节点此时已故障，不能再接受操作指令），

客户端可以订阅 +switch-master频道，一旦 「Redis Sentinel」 结束了对主节点的故障转移就会发布主节点的的消息。

redission 分布式锁

redission 开源框架提供一些便捷操作 Redis 的方法，其中比较出名的 redission 基于 Redis 的实现分布式锁。

今天我们来看下 Redis 的实现分布式锁中如何使用 Redis 发布订阅机制，提高加锁的性能。

首先我们来看下 redission 加锁的方法：

Redisson redisson = ....
RLock redissonLock = redisson.getLock("xxxx");
redissonLock.lock();

RLock 继承自 Java 标准的 Lock 接口,调用 lock 方法,如果当前锁已被其他客户端获取，那么当前加锁的线程将会被阻塞，直到其他客户端释放这把锁。

这里其实有个问题，当前阻塞的线程如何感知分布式锁已被释放呢？

这里其实有两种实现方法：

第一钟，定时查询分布时锁的状态，一旦查到锁已被释放（Redis 中不存在这个键值），那么就去加锁。

实现伪码如下：

while (true) {
  boolean result=lock();
  if (!result) {
    Thread.sleep(N);
  }
}

这种方式实现起来起来简单，不过缺点也比较多。

如果定时任务时间过短，将会导致查询次数过多，其实这些都是无效查询。

如果定时任务休眠时间过长，那又会导致加锁时间过长，导致加锁性能不好。

那么第二种实现方案，就是采用服务通知的机制，当分布式锁被释放之后，客户端可以收到锁释放的消息，然后第一时间再去加锁。

这个服务通知的机制我们可以使用 Redis 发布订阅模式。

当线程加锁失败之后，线程将会订阅 redisson_lock__channel_xxx（xx 代表锁的名称）频道，使用异步线程监听消息，然后利用 Java 中 Semaphore 使当前线程进入阻塞。

一旦其他客户端进行解锁，redission 就会往这个redisson_lock__channel_xxx 发送解锁消息。

等异步线程收到消息，将会调用 Semaphore 释放信号量，从而让当前被阻塞的线程唤醒去加锁。

❝
ps：这里只是简单描述了 redission 加锁部分原理，出于篇幅，这里就不再消息解析源码。

感兴趣的小伙伴可以自己看下 redission 加锁的源码。
❞

通过发布订阅机制，被阻塞的线程可以及时被唤醒，减少无效的空转的查询，有效的提高的加锁的效率。

❝
ps: 这种方式，性能确实提高，但是实现起来的复杂度也很高，这部分源码有点东西，快看晕了。

5 实时聊天应用示例🍀

场景描述

假设你正在开发一个实时聊天应用，用户可以在不同的聊天室中进行交流。每个聊天室都有多个用户，当一个用户发送消息时，其他所有在该聊天室中的用户都应该立即收到这条消息。

传统解决方案的问题

轮询（Polling）：客户端定期向服务器发送请求，检查是否有新消息。这种方式效率低下，且响应时间较长。
长轮询（Long Polling）：客户端发送请求后，服务器保持连接打开，直到有新消息或超时。虽然比普通轮询更高效，但仍存在延迟和资源浪费问题。
WebSocket：使用WebSocket可以实现全双工通信，但设置和维护相对复杂，并且需要处理更多的网络层细节。

使用Redis Pub/Sub的优势

实时性：Redis的Pub/Sub机制可以实现实时的消息传递，确保消息几乎即时送达。
简单易用：Redis的Pub/Sub命令简单直观，易于集成到现有系统中。
高效率：Redis是内存数据库，消息传递速度快，适合高频次的消息传输。
灵活性：可以轻松地扩展到多个聊天室和大量用户。

示例实现

初始化Redis客户端
在服务器端和客户端分别初始化Redis客户端
创建聊天室频道
每个聊天室对应一个Redis频道。例如，chatroom:1 表示第一个聊天室
客户端订阅频道
客户端在进入某个聊天室时，订阅对应的频道
客户端发布消息
当用户发送消息时，客户端将消息发布到对应的频道
服务器端管理聊天室
服务器端可以负责管理和验证用户的订阅和发布操作，确保只有合法用户能够参与聊天

# 相关代码如下: 
---------- 聊天室 服务器 代码 如下 ------------
#! /usr/bin/python
# -*- coding: UTF-8 -*-
import redis

# 初始化Redis客户端
r = redis.Redis(host='localhost', port=6379, db=0, password='kdlrdsha2019')

# 订阅聊天室列表
pubsub = r.pubsub()
pubsub.subscribe('chatroom:1')

for message in pubsub.listen():
    if message['type'] == 'message':
        print(f"Received message: {message['data'].decode('utf-8')}")

---------- 客户端使用 代码如下 ----------------
[root@test ~]# python3
Python 3.12.0 (main, Oct 27 2023, 14:11:11) [GCC 4.8.5 20150623 (Red Hat 4.8.5-44)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import redis
>>> r = redis.Redis(host='localhost', port=6379, db=0, password='kdlrdsha2019')
>>> r.publish('chatroom:1', 'Hello, everyone!')
1

---------- 服务端返回如下 ------------------
[root@test Redis]# python3 Redis_Pub_Sub.py
Received message: Hello, everyone!

代码示例

以下是Python示例代码，展示了如何使用Redis的Pub/Sub功能实现简单的实时聊天应用。

#! /usr/bin/python
# -*- coding: UTF-8 -*-
import redis
import threading

# 初始化Redis客户端, Redis服务器有密码的话需要带password参数
r = redis.Redis(host='localhost', port=6379, db=0)

def subscribe_to_chatroom(chatroom):
    pubsub = r.pubsub()
    pubsub.subscribe(chatroom)
    
    print(f"Subscribed to {chatroom}")
    for message in pubsub.listen():
        if message['type'] == 'message':
            print(f"Received message in {chatroom}: {message['data'].decode('utf-8')}")

def send_message(chatroom, message):
    r.publish(chatroom, message)
    print(f"Message sent to {chatroom}: {message}")

# 创建两个线程，模拟两个客户端订阅同一个聊天室
thread1 = threading.Thread(target=subscribe_to_chatroom, args=('chatroom:1',))
thread2 = threading.Thread(target=subscribe_to_chatroom, args=('chatroom:1',))

thread1.start()
thread2.start()

# 模拟发送消息
send_message('chatroom:1', 'Hello, everyone!')
send_message('chatroom:1', 'How are you?')