答: redis
发布订阅是一种消息通信通信模式,由发送者(pub)
发送消息,订阅者(sub)
接收消息。
如下图client2、4、5
就是订阅着,订阅了channel1
的消息。
当channel1
要发送消息时,这几个订阅者都会实时收到消息。
答: 有两种,一种基于频道的,还有一种是基于匹配的模式:
指令如下所示,可以看到订阅者可以订阅多个频道
subscribe channel [channel ...]
所以我们开启一个redis客户端,订阅一个channel:sport
的频道
# 客户端1 订阅 channel:sport
127.0.0.1:6379> SUBSCRIBE channel:sport
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "channel:sport"
3) (integer) 1
发布消息的指令格式为:
publish channel message
此时,我们再开启一个redis
客户端,发布一条消息到channel:sport
的频道
# 另一个客户端发送消息
127.0.0.1:6379> PUBLISH channel:sport "this is why we play"
(integer) 1
127.0.0.1:6379>
此时刚刚订阅消息的redis
客户端就会实时的收到这条消息
127.0.0.1:6379> SUBSCRIBE channel:sport
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "channel:sport"
3) (integer) 1
# 如下便是实时收到的消息内容
1) "message"
2) "channel:sport" #频道
3) "this is why we play" # 消息内容
有时候我们会订阅多个频道,我们不可能每次都去手动增加订阅的频道,例如我们当前订阅频道有:c1、c2、c3、c4、c5、c6、c7、c8
。将来还可能出现c9等情况。我们不可能实时去添加订阅的频道。
观察上面的频道我们发现频道都是以c开头,后续的数字不断变化,所以我们完全可以使用模式匹配来实现频道订阅。
模式订阅和取消的命令为
psubscribe pattern [pattern...]
punsubscribe [pattern [pattern ...]]
关于parttern
常见的匹配符有
1. *:表示任意占位符,例如c*,可以匹配c、c1、c111
2. ?*:匹配一个及以上个占位符
3. ?:表示匹配一个占位符
我们希望订阅c1-c9
的频道基于模式匹配
我们就能够做到这一点
我们首先开启一个客户端,使用模式匹配发起订阅
# 订阅匹配cxx相关的模式
PSUBSCRIBE c?*
然后我们在开启另一个客户端,发送消息到c1频道
127.0.0.1:6379> PUBLISH c1 "this is c1 message"
(integer) 1
127.0.0.1:6379>
刚刚订阅的客户端就会收到消息
127.0.0.1:6379> PSUBSCRIBE c?*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "c?*"
3) (integer) 1
1) "pmessage"
2) "c?*"
3) "c1"
4) "this is c1 message"
注意:当你订阅PSUBSCRIBE c?* c1
订阅时,若另一个客户端发送消息到c1你会收到两条消息(原因会在后文源码解析时补充)。
如下便是PSUBSCRIBE c?* c1
的收到PUBLISH c1 "this is c1 message"的消息内容
127.0.0.1:6379> PSUBSCRIBE c?* c1
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "c?*"
3) (integer) 1
1) "psubscribe"
2) "c1"
3) (integer) 2
1) "pmessage"
2) "c1"
3) "c1"
4) "this is c1 message"
1) "pmessage"
2) "c?*"
3) "c1"
4) "this is c1 message"
答: 使用场景如:聊天室、公告牌等需要实现消息解耦的场景都可以使用消息订阅发布,如下便是一个基于redis实现视频订阅的管理系统。
而优缺点也很明显:
缺点: 相较于Kafka
、RocketMQ
,Redis
的实现略显粗糙,无法实现消息堆积与回溯
优点: 实现简单,若不需要消息回溯等需求使用redis
绰绰有余
答: 我们可以从源码的角度分析分析频道模式的发布与订阅:
频道发布订阅是基于字典这个数据结果,以频道作为key,用链表作为value存储所有订阅该频道的当订阅者,如下图,当channel1
有消息时,程序就通过channel1
订阅链,逐个给订阅者发送消息。
当如果客户端 client10086
执行命令 SUBSCRIBE channel1 channel2 channel3
时,程序就会将这个订阅者挂到对应三个频道中。
同理,订阅者通过UNSUBSCRIBE
退订频道,程序也是通过pubsub_channels
找到对应的链表将这个节点移除。
说完了频道的,我们再来说说redis
匹配模式发布与订阅:
与频道订阅有所不同,匹配模式订阅数据结构如下所示,可以看到client123、client256
订阅了tweet.shop.*
,订阅者的数据结构是一个记录客户端以及客户端订阅的模式频道的结构体,从源码我们可以看到这个结构体是这样的:
typedef struct pubsubPattern {
redisClient *client; //指向订阅者
robj *pattern;//指向匹配模式
} pubsubPattern;
从上图我们也可以看到redisServer
就是记录一个个客户端匹配模式的结构体,从源码我们就可以看到,它包含了一个pubsub_patterns
的指针用于记录每个模式订阅者
struct redisServer {
// ...
list *pubsub_patterns;
// ...
};
例如client10086
订阅broadcast.live.*
,程序就会在redisServer
所指向的链表中挂一个新的pubsub_patterns
,如下图所示:
我们希望一个用户订阅channel:sport
,一个用户订阅channel:stock,而另一个用户两个都订阅
首先pom配置引入相关依赖
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
spring boot的application.yml配置redis配置
spring:
redis:
host: 127.0.0.1
port: 6379
password:
database: 0
package com.example.demo;
import org.springframework.stereotype.Component;
/**
* 发布消息接口类
*/
@Component
public interface RedisMsg {
void receiveMessage(String message);
}
/**
* 订阅所有channel:*的消息
*/
public class RedisSubPChannel implements RedisMsg {
public void receiveMessage(String message) {
System.out.println("RedisSubPChannel收到消息消息" + message);
}
}
package com.example.demo;
/**
* 订阅所有channel:sport最新消息
*/
public class RedisSubSport implements RedisMsg{
public void receiveMessage(String message){
System.out.println("RedisSubSport获取体育频道最新消息:"+message);
}
}
package com.example.demo;
/**
* 订阅所有channel:stock消息
*/
public class RedisSubStock implements RedisMsg {
@Override
public void receiveMessage(String message) {
System.out.println("RedisSubStock获取channel:stock最新消息:"+message);
}
}
@Configuration
@EnableCaching
public class RedisConfig{
/**
* Redis消息监听器容器
* @param connectionFactory
* @return
*/
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
//订阅三个频道
container.addMessageListener(listenerAdapter(new RedisSubSport()),new PatternTopic("channel:sport"));
container.addMessageListener(listenerAdapter(new RedisSubStock()),new PatternTopic("channel:stock"));
container.addMessageListener(listenerAdapter(new RedisSubPChannel()),new PatternTopic("channel:*"));
return container;
}
/**
* 配置消息接收处理类
* @param redisMsg 自定义消息接收类
* @return
*/
@Bean()
@Scope("prototype")
MessageListenerAdapter listenerAdapter(RedisMsg redisMsg) {
//这个地方 是给messageListenerAdapter 传入一个消息接受的处理器,利用反射的方法调用“receiveMessage”
//也有好几个重载方法,这边默认调用处理器的方法 叫handleMessage 可以自己到源码里面看
return new MessageListenerAdapter(redisMsg, "receiveMessage");//注意2个通道调用的方法都要为receiveMessage
}
}
//定时器
@EnableScheduling
@Component
public class SenderController {
@Autowired
private StringRedisTemplate stringRedisTemplate;
//向redis消息队列index通道发布消息
@Scheduled(fixedRate = 2000)
public void sendMessage() {
stringRedisTemplate.convertAndSend("channel:stock", "股票涨了"+Math.random()+"个百分点");
stringRedisTemplate.convertAndSend("channel:sport", "湖人队 No."+((int)(Math.random()*100)+1)+"选手得分");
}
}
RedisSubStock获取channel:stock最新消息:股票涨了0.46015187165884985个百分点
RedisSubPChannel收到消息消息股票涨了0.46015187165884985个百分点
RedisSubSport获取体育频道最新消息:湖人队 No.40选手得分
RedisSubPChannel收到消息消息湖人队 No.40选手得分
RedisSubStock获取channel:stock最新消息:股票涨了0.5293839028647649个百分点
RedisSubPChannel收到消息消息股票涨了0.5293839028647649个百分点
RedisSubSport获取体育频道最新消息:湖人队 No.26选手得分
RedisSubPChannel收到消息消息湖人队 No.26选手得分
RedisSubPChannel收到消息消息股票涨了0.6779431428770823个百分点
RedisSubStock获取channel:stock最新消息:股票涨了0.6779431428770823个百分点
RedisSubSport获取体育频道最新消息:湖人队 No.35选手得分
RedisSubPChannel收到消息消息湖人队 No.35选手得分
RedisSubStock获取channel:stock最新消息:股票涨了0.40332150203704个百分点
RedisSubPChannel收到消息消息股票涨了0.40332150203704个百分点
RedisSubSport获取体育频道最新消息:湖人队 No.14选手得分
RedisSubPChannel收到消息消息湖人队 No.14选手得分
RedisSubStock获取channel:stock最新消息:股票涨了0.8407461728769783个百分点
RedisSubPChannel收到消息消息股票涨了0.8407461728769783个百分点
RedisSubSport获取体育频道最新消息:湖人队 No.25选手得分
RedisSubPChannel收到消息消息湖人队 No.25选手得分