Redis 发布订阅

发布订阅：消息发布者发布消息 和 消息订阅者接收消息，两者之间通过某种媒介联系起来

例如订杂志，当自己订阅了爱格杂志，每个月会发刊一本。到发布的时候派送员将杂志送到自己手上就能看到杂志内容。只有我们订阅了该杂志才会派送给我们

Redis 发布订阅（pub/sub）是一种 消息通信模式 ：发送者（pub）发送消息，订阅者（sub）接收消息。

Redis 客户端可以订阅任意数量的频道。

订阅 / 发布消息图：

图中可以看出，所需：

1. 消息发送者 、 2. 频道 、 3. 消息订阅者

发布订阅机制

1. 当一个客户端通过 PUBLISH 命令向订阅者发布消息的时候，称这个客户端为发布者publisher
2. 当一个客户端通过subscribe 或者 PSUBSCRIBE 接收消息时，称这个客户端为 订阅者 subscriber
3. 为了解耦发布者和订阅者之间的关系，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

1. Listener模式

创建一个监听容器

@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","测试：消息发布者发布消息");    }}

测试结果

2. 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