千家信息网

请输入关键字词

热门搜索排行

最新搜索排行

导航： 首页 > 开发技术 >

SpringBoot Redis发布订阅模式的方法是什么

发表于：2025-01-19 作者：千家信息网编辑
千家信息网最后更新 2025年01月19日，本篇内容主要讲解"SpringBoot Redis发布订阅模式的方法是什么"，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习"SpringBoot Redis发
千家信息网最后更新 2025年01月19日SpringBoot Redis发布订阅模式的方法是什么

本篇内容主要讲解"SpringBoot Redis发布订阅模式的方法是什么"，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习"SpringBoot Redis发布订阅模式的方法是什么"吧!

注意：redis的发布订阅模式不可以将消息进行持久化，订阅者发生网络断开、宕机等可能导致错过消息。

Redis命令行下使用发布订阅

publish 发布

发布者通过以下命令可以往指定channel发布message

redis> publish channel message

subscribe 订阅

订阅者通过以下命令可以订阅一个或多个频道，如果频道不存在则会创建

redis> subscribe channel [channel ...]

对于redis的发布订阅的命令就这么简单。那么接下来我们在springboot中如何使用发布订阅的功能呢？

SpringBoot中使用Redis的发布订阅功能

添加依赖配置redis信息和连接池什么的就不说了，如果添加的有commons-pool2依赖的话，会自动帮我们配置redis连接池的

发布者

相对于订阅者来说，发布者的实现方式很简单，以下方式就可以往channel中发送message了。

@Resource    private RedisTemplate redisTemplate;    public void publish(){    // 使用高级的redisTemplate    redisTemplate.convertAndSend("channel","message");        // 使用低级的connection 实际上redisTemplate的底层就是使用的下面的方式    redisTemplate.execute(new RedisCallback() {          @Override          public Object doInRedis(RedisConnection connection) throws DataAccessException {              connection.publish("channel".getBytes(StandardCharsets.UTF_8), "message".getBytes(StandardCharsets.UTF_8));              return null;         }     }, true);     // true这个参数意思是 是否将redis连接暴露给回调代码，大多数情况下设置true就可以了，往后深入的话可以看到     RedisConnection connToExpose = (exposeConnection ? connToUse : createRedisConnectionProxy(connToUse)); 如果为false的话会创建redis连接的代理}
订阅者
订阅者因为涉及到连接、线程等 所以内容相对会多一点
@Resource    private RedisTemplate redisTemplate;    public void subscribe() {        redisTemplate.execute(new RedisCallback() {            @Override            public Object doInRedis(RedisConnection connection) throws DataAccessException {                // 我定义了一个全局的 ConcurrentHashMap 用来存放连接 因为后面的取消订阅的线程要和订阅的线程用同一个连接                map.put("connection",connection);                // subscribe 按频道订阅 该方法会阻塞该线程 只有取消订阅才会释放该线程                connection.subscribe(new MessageListener() {                    @Override                    public void onMessage(Message message, byte[] pattern) {                        log.info("接收到消息");                        System.out.println(new String(message.getBody()));                    }                }, "channelOne".getBytes(StandardCharsets.UTF_8), "channelTwo".getBytes(StandardCharsets.UTF_8));                // 按模式订阅 pSubscribe 只有取消订阅才会释放该线程//                connection.pSubscribe(new MessageListener() {//                    @Override//                    public void onMessage(Message message, byte[] pattern) {//                        System.out.println(new String(message.getBody()));//                    }//                }, "patternOne".getBytes(StandardCharsets.UTF_8), "patternOne".getBytes(StandardCharsets.UTF_8));                return null;            }        }, true);    }
如何取消订阅呢？从刚才的map里取到连接
RedisConnection the = map.get("connection");    Subscription subscription = the.getSubscription();    subscription.unsubscribe();
消息监听容器
上面的那种订阅为低级订阅，由于连接在调用subscribe的时候会导致当前线程阻塞，这种方式需要对每个监听器连接和线程管理，所以spring提供了RedisMessageListenerContainer类来帮我们完成这些工作。
RedisMessageListenerContainer顾名思义可以知道它是一个消息监听容器
详情请参考官方文档
如何实现
@Configurationpublic class DefaultMessageListenerContainerConfig {    @Bean    public RedisMessageListenerContainer container(RedisConnectionFactory factory) {        RedisMessageListenerContainer container = new RedisMessageListenerContainer();        container.setConnectionFactory(factory);        // 官方推荐我们使用自定义的线程池或者使用TaskExecutor        container.setTaskExecutor(executor());        container.addMessageListener(new MessageListener() {            @Override            public void onMessage(Message message, byte[] pattern) {                System.out.println(Thread.currentThread().getName() + ": " + new String(message.getBody()));            }        }, new ChannelTopic("message"));        return container;    }    @Bean    public TaskExecutor executor() {        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();        executor.setCorePoolSize(Runtime.getRuntime().availableProcessors());        executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 2);        executor.setQueueCapacity(100);        executor.initialize();        return executor;    }}
这个时候我们在redis命令行内使用 publish channel message 的时候，我们的spring程序就可以订阅到消息了。
再说下 MessageListenerAdapter
我们可以通过 MessageListenerAdapter 消息接收者包装进去，消息接收者不会和redis有任何耦合。
官方文档给了spring传统的xml的方式配置的，下面我给出基于configuration配置的代码
public interface MessageDelegate {    void handleMessage(String message);}public class DefaultMessageDelegate implements MessageDelegate {    @Override    public void handleMessage(String message) {        System.out.println(message);    }}@Configurationpublic class MessageListenerContainerConfig {    @Autowired    private DefaultMessageDelegate defaultMessageDelegate;    @Bean    public RedisMessageListenerContainer container(RedisConnectionFactory factory,                                                   MessageListenerAdapter messageListenerAdapter) {        RedisMessageListenerContainer container = new RedisMessageListenerContainer();        container.setConnectionFactory(factory);        container.setTaskExecutor(executor());        Map> map = new HashMap<>();        List channelTopics = new ArrayList<>();        ChannelTopic channelTopic = new ChannelTopic("message");        channelTopics.add(channelTopic);        map.put(messageListenerAdapter, channelTopics);        container.setMessageListeners(map);        return container;    }    @Bean    public TaskExecutor executor() {        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();        executor.setCorePoolSize(Runtime.getRuntime().availableProcessors());        executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 2);        executor.setQueueCapacity(100);        executor.initialize();        return executor;    }    @Bean    public MessageListenerAdapter messageListenerAdapter() {        // handleMessage 参数消息来的时候要调用的方法 默认是 handleMessage        return new MessageListenerAdapter(defaultMessageDelegate, "handleMessage");    }}
如果我们要在程序运行时添加订阅或者取消订阅的时候该怎么办呢？
我们需要提前准备好消息侦听器，添加的时候把侦听器注入到消息容器
取消的时候就调用消息容器的remove方法把侦听器删除掉即可。
到此，相信大家对"SpringBoot Redis发布订阅模式的方法是什么"有了更深的了解，不妨来实际操作一番吧！这里是网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！


        


        
 很赞哦！ 

        


        



        订阅
            消息
            线程
            方法
            时候
            模式
            命令
            方式
            订阅者
            容器
            频道
            配置
            侦听器
            内容
            发布者
            官方
            监听
            低级
            代码
            会创
    
数据库的安全要保护哪些东西
数据库安全各自的含义是什么
生产安全数据库录入
数据库的安全性及管理
数据库安全策略包含哪些
海淀数据库安全审计系统
建立农村房屋安全信息数据库
易用的数据库客户端支持安全管理
连接数据库失败ssl安全错误
数据库的锁怎样保障安全

贵州超级服务器虚拟主机
数据库视图 dao
中国能造出全套的网络技术吗
数据库企业版
速达v3数据库日志已满
博彦科技互联网内容
北京纵横网络技术有限公司招聘
数据库关系号
刷步道乐跑服务器
网络安全黑板报布置书桌
大连教育局网络安全宣传周
负载均衡服务器
建立数据库表的单词命令
服务器虚拟化软件管理系统
服务器在国外被限流怎么办
教育部网络安全7
软件开发授权开发模板
佛山微商软件开发市场价
steam恐龙岛都要啥服务器
全球数据库行业
中国台湾java软件开发需求
数据库备份的意义
网络安全盾牌彩色简笔画
网络技术工资大约多少
浅谈办公自动化网络安全论文大纲
仓库进出库 数据库
思科网络技术度量是什么意思
央企 软件开发
我和网络安全作文
数据库dateto
        

        


        
        

        

          
        

      
    


    



    

      

        
相关文章

      

      
    






   





  



































0