Redis 发布订阅
发布订阅：消息发布者发布消息 和 消息订阅者接收消息，两者之间通过某种媒介联系起来
例如订杂志，当自己订阅了爱格杂志，每个月会发刊一本。到发布的时候派送员将杂志送到自己手上就能看到杂志内容。只有我们订阅了该杂志才会派送给我们
Redis 发布订阅（pub/sub）是一种 消息通信模式 ：发送者（pub）发送消息，订阅者（sub）接收消息。
Redis 客户端可以订阅任意数量的频道。
订阅 / 发布消息图：
图中可以看出，所需：
- 消息发送者 、 2. 频道 、 3. 消息订阅者
发布订阅机制
- 当一个客户端通过
PUBLISH命令向订阅者发布消息的时候，称这个客户端为发布者publisher - 当一个客户端通过
subscribe或者PSUBSCRIBE接收消息时，称这个客户端为 订阅者subscriber - 为了解耦发布者和订阅者之间的关系，Redis 使用了频道
channel(频道)作为两者之间的中介，发布者直接把消息发送给 channel，而 channel 负责把消息发送给订阅者，发布者和订阅者之间没有直接的联系，都不知道对方的存在
订阅者 1，2，3 订阅了频道 channel，当有消息发布给频道时，这个消息就会被发送到三个订阅者客户端
demo 实现
引入依赖
<dependency> <groupId>org.springframework.bootgroupId> <artifactId>spring-boot-starter-data-redisartifactId>dependency>添加配置文件
spring: redis: host: 127.0.0.1 database: 5 password: port: 6379创建一个监听容器
@Configurationpublic class CatListenerConfig extends CachingConfigurerSupport { @Bean RedisMessageListenerContainer container(RedisConnectionFactory factory){ RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(factory); //订阅一个通道 该处的通道名是发布消息时的名称 container.setConnectionFactory(connectionFactory); //订阅了一个叫cat 的通道 container.addMessageListener(catAdapter, new PatternTopic("cat")); container.addMessageListener(fishAdapter, new PatternTopic("fish")); return container; } }@Componentpublic class CatListener implements MessageListener { @Override public void onMessage(Message message, byte[] bytes) { System.out.println("我是监听者,我监听到的消息是 " + message.toString()); }}package com.maoxs.listener;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.data.redis.connection.Message;import org.springframework.data.redis.connection.MessageListener;import org.springframework.data.redis.core.RedisTemplate;public class FishListener implements MessageListener { @Autowired RedisTemplate redisTemplate; @Override public void onMessage(Message message, byte[] bytes) { System.out.println("我是Fish监听" + message.toString()); }}建测试类，测试发布监听
@RestControllerpublic class TestController { @Resource StringRedisTemplate stringRedisTemplate; @PostMapping("/cat") public void test2(){ stringRedisTemplate.convertAndSend("cat","测试：消息发布者发布消息"); } @PostMapping("/fish") public void fish(){ stringRedisTemplate.convertAndSend("fish","测试：消息发布者发布消息"); }}测试结果
-
Adapter模式
1 配置
@Bean public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,Map<String, MessageListenerAdapter> adapterMap) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); adapterMap.keySet().forEach(topic-> container.addMessageListener(adapterMap.get(topic),new PatternTopic(topic))); return container; }监听1
package com.sst.loan.risk.listener;import com.sst.loan.risk.manager.RuleLoadManager;import lombok.extern.slf4j.Slf4j;import org.springframework.data.redis.connection.Message;import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;import org.springframework.stereotype.Component;import javax.annotation.Resource;@Slf4j@Component("ruleRefreshAdapter")public class RuleRefreshAdapter extends MessageListenerAdapter { @Override public void onMessage(Message message, byte[] bytes) { log.info(">>>>>>> 消息适配器收到刷新规则的请求 <<<<<<<<<<<<<<"); }}监听2
package com.sst.loan.risk.listener;import com.sst.loan.risk.manager.RuleLoadManager;import lombok.extern.slf4j.Slf4j;import org.springframework.data.redis.connection.Message;import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;import org.springframework.stereotype.Component;import javax.annotation.Resource;@Slf4j@Component("ruleRefreshAdapter2")public class RuleRefreshAdapter2 extends MessageListenerAdapter { @Override public void onMessage(Message message, byte[] bytes) { log.info(">>>>>>> 消息适配器收到刷新规则的请求 <<<<<<<<<<<<<<"); }}测试
@Resource private StringRedisTemplate stringRedisTemplate; @GetMapping("refresh") public Object refresh() { stringRedisTemplate.convertAndSend("ruleRefreshAdapter","refresh"); return Result.success(); } @GetMapping("refresh") public Object refresh() { stringRedisTemplate.convertAndSend("ruleRefreshAdapter2","refresh"); return Result.success(); }
来源地址：https://blog.csdn.net/caidingnu/article/details/131184710
