前言
Redis除了可以用作缓存数据外,另一个重要用途是它实现了发布订阅(pub/sub)消息通信模式:发送者(pub)发送消息,订阅者(sub)接收消息。
为了实现redis的发布订阅机制,首先要打开redis服务;其次,引入redis需要的jar包,在pom.xml配置文件加入以下代码:
<dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>2.1.0</version> </dependency>
由于订阅消息通道需要再tomcat启动时触发,因此,需要创建一个listener监听器,在监听器里实现redis订阅,在web.xml里配置监听器如下:
<listener> <listener-class>com.test.listener.InitListener</listener-class> </listener>
一、订阅消息(InitListener实现)
redis支持多通道订阅,一个客户端可以同时订阅多个消息通道,如下代码所示,订阅了13个通道。由于订阅机制是线程阻塞的,需要额外开启一个线程专门用于处理订阅消息及接收消息处理。
public class InitListener implements ServletContextListener{ private Logger logger = Logger.getLogger(InitListener.class); @Override public void contextInitialized(ServletContextEvent sce) { logger.info("启动tomcat");// 连接redis Map<String, String> proMap = PropertyReader.getProperties(); final String url = proMap.get("redis.host"); final Integer port = Integer.parseInt(proMap.get("redis.port")); final ClassPathXmlApplicationContext classPathXmlApplicationContext = new ClassPathXmlApplicationContext("classpath*:applicationContext.xml"); final RedisSubListener redisSubListener = (RedisSubListener) classPathXmlApplicationContext.getBean("redisSubListener"); // 为防止阻塞tomcat启动,开启线程执行 new Thread(new Runnable(){ public void run(){ // 连接redis,建立监听 Jedis jedis = null; while(true){ //解码资源更新通知,画面选看回复,画面选看停止回复,预案启动,预案停止,轮切启动,轮切停止,预案启动回复,预案停止回复,轮切启动回复,轮切停止回复,监视屏分屏状态通知,画面状态通知 String[] channels = new String[] { "decodeResourceUpdateNtf", "tvSplitPlayRsp","tvSplitPlayStopRsp", "planStartStatusNtf", "planStopStatusNtf", "pollStartStatusNtf", "pollStopStatusNtf", "planStartRsp","planStopRsp","pollStartRsp","pollStopRsp","tvSplitTypeNtf","tvSplitStatusNtf"}; try{ jedis = new Jedis(url,port); logger.info("redis请求订阅通道"); jedis.subscribe(redisSubListener,channels); logger.info("redis订阅结束"); }catch(JedisConnectionException e){ logger.error("Jedis连接异常,异常信息 :" + e); }catch(IllegalStateException e){ logger.error("Jedis异常,异常信息 :" + e); } try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } if(jedis != null){ jedis = null; } } }}) .start(); }
最后在spring配置文件里接入以下配置:
<!-- redis --> <bean id="redisMessageService" class="com.test.service.impl.RedisMessageServiceImpl" scope="singleton"> <property name="webSocketService"><ref local="webSocketService" /></property> <property name="tvSplitStatusDao" ref="tvSplitStatusDao"></property> </bean> <bean id="redisSubListener" class="com.test.common.RedisSubListener" scope="singleton"> <property name="redisMessageService"><ref local="redisMessageService" /></property> </bean>
RedisMessageServiceImpl用于处理接收的redis消息。
二、发布消息
public class RedisPublishUtil { private Logger logger = Logger.getLogger(RedisPublishUtil.class); public static Jedis pubJedis; private static Map<String, String> proMap = PropertyReader.getProperties(); private static final String redisPort = proMap.get("redis.port"); private static String url = proMap.get("redis.host"); private static final int port = Integer.parseInt(redisPort); public void setPubJedis(Jedis jedis) { RedisPublishUtil.pubJedis = jedis; } public Jedis getPubJedis() { if (pubJedis == null) { createJedisConnect(); } // 返回对象 return pubJedis; } public Jedis createJedisConnect(){ // 连接redis logger.info("===创建连接jedis====="); try { pubJedis = new Jedis(url, port); } catch (JedisConnectionException e) { logger.error("Jedis连接异常,异常信息 :" + e.getMessage()); try { Thread.sleep(1000); logger.info("发起重新连接jedis"); createJedisConnect(); } catch (InterruptedException except) { except.printStackTrace(); } } // 返回对象 return pubJedis; } //公共发布接口 public void pubRedisMsg(String msgType,String msg){ logger.info("redis准备发布消息内容:" + msg); try { this.getPubJedis().publish(msgType, msg); } catch (JedisConnectionException e) { logger.error("redis发布消息失败!", e); this.setPubJedis(null); logger.info("重新发布消息,channel="+msgType); pubRedisMsg(msgType, msg); } } }
public class PropertyReader { private static Logger logger = Logger.getLogger(PropertyReader.class); /* * 获得数据库链接的配置文件 */ public static Map<String,String> getProperties(){ logger.info("读取redis配置文件开始。。。"); Properties prop = new Properties(); Map<String,String> proMap = new HashMap<String,String>(); try { //读取属性文件redis.properties InputStream in= PropertyReader.class.getClassLoader().getResourceAsStream("redis.properties"); prop.load(in); ///加载属性列表 Iterator<String> it=prop.stringPropertyNames().iterator(); while(it.hasNext()){ String key=it.next(); proMap.put(key, prop.getProperty(key)); } in.close(); logger.info("读取redis配置文件成功。。。"); } catch (Exception e) { logger.error("读取redis配置文件异常!", e); e.printStackTrace(); } return proMap; } }
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持小牛知识库。
本文向大家介绍Spring boot+redis实现消息发布与订阅的代码,包括了Spring boot+redis实现消息发布与订阅的代码的使用技巧和注意事项,需要的朋友参考一下 一.创建spring boot项目 二.编辑yml配置文件 三.配置Redis 四.编写RedisUtil消息发布方法 五.配置消息监听 六.消息订阅方法 七.定时消息发布测试 八.测试结果 九.发布对象User实体 到
本文向大家介绍Vue发布订阅模式实现过程图解,包括了Vue发布订阅模式实现过程图解的使用技巧和注意事项,需要的朋友参考一下 vue项目中不同组件间通信一般使用vuex,通常情况下vuex和EventBus不应该混用,不过某些场景下不同组件间只有消息的交互,这时使用EventBus消息通知的方式就更合适一些。 图解 html Dvue.js 以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大
主要内容:发布/订阅流程,常用命令汇总,基本命令应用Redis PubSub 模块又称发布订阅者模式,是一种消息传递系统,实现了消息多播功能。发布者(即发送方)发送消息,订阅者(即接收方)接收消息,而用来传递消息的链路则被称为 channel。在 Redis 中,一个客户端可以订阅任意数量的 channel(可译为频道)。 消息多播:生产者生产一次消息,中间件负责将消息复制到多个消息队列中,每个消息队列由相应的消费组进行消费,这是分布式系统常用的
本文向大家介绍Redis 订阅发布_Jedis实现方法,包括了Redis 订阅发布_Jedis实现方法的使用技巧和注意事项,需要的朋友参考一下 我想到使用Redis的订阅发布模式是用来解决推送问题的~。 对于概念性的叙述,多多少少还是要提一下的: 什么是Redis发布订阅?Redis发布订阅是一种消息通信模式,发送者通过通道A发送消息message,订阅过通道A的客户端就可以接收到消息messag
本文向大家介绍Redis发布订阅和实现.NET客户端详解,包括了Redis发布订阅和实现.NET客户端详解的使用技巧和注意事项,需要的朋友参考一下 前言 发布订阅在设计模式中也可以说是观察者模式,针对这个模式是处理对象间一对多的依赖关系的,当一个对象发生变化,其它依赖他的对象都要得到通知并更新。 然而它也有自己的缺点,就是当主题发生一系列的变化时,观察者都要做批量的更新,如果这样的更新成本很高,那
本文向大家介绍nodejs redis 发布订阅机制封装实现方法及实例代码,包括了nodejs redis 发布订阅机制封装实现方法及实例代码的使用技巧和注意事项,需要的朋友参考一下 nodejs redis 发布订阅机制封装 最近项目使用redis,对publish 和 subscribe的使用进行了了解,并进行了封装。 然后通过exports.pubsub=new PubSub() 将