redis发布订阅Java代码实现过程解析

 更新时间:2019年09月17日 11:32:31   作者:冰湖一角  
这篇文章主要介绍了redis发布订阅Java代码实现过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下

前言

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;
  }
}

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。

相关文章

  • Java中ExecutorService和ThreadPoolExecutor运行原理

    Java中ExecutorService和ThreadPoolExecutor运行原理

    本文主要介绍了Java中ExecutorService和ThreadPoolExecutor运行原理,文中通过示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2021-08-08
  • Java8 日期和时间类的基本使用

    Java8 日期和时间类的基本使用

    这篇文章主要介绍了Java 日期和时间类的基本使用,帮助大家更好的理解和学习Java,感兴趣的朋友可以了解下
    2020-08-08
  • Java并发编程之threadLocal

    Java并发编程之threadLocal

    ThreadLocal是JDK包提供的,它提供了线程本地变量,也就是说如果创建了一个ThreadLocal变量,需要的朋友可以参考一下哟
    2021-09-09
  • Jmeter逻辑控制器事务控制器使用方法解析

    Jmeter逻辑控制器事务控制器使用方法解析

    这篇文章主要介绍了Jmeter逻辑控制器事务控制器使用方法解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-11-11
  • 解析Nacos的API居然存在这么严重的漏洞

    解析Nacos的API居然存在这么严重的漏洞

    这篇文章主要介绍了Nacos的API居然存在这么严重的漏洞,Nacos为我们提供了大量API,但是这些API默认是没有开启认证的,直接可以访问,针对于这一点我们也都可以去验证一下,本文给大家详细讲解,感兴趣的朋友跟随小编一起看看吧
    2022-09-09
  • Java Socket实现文件传输示例代码

    Java Socket实现文件传输示例代码

    这篇文章主要介绍了Java Socket实现文件传输示例代码,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2017-01-01
  • SpringBoot优雅接收前端请求参数的详细过程

    SpringBoot优雅接收前端请求参数的详细过程

    这篇文章主要介绍了SpringBoot如何优雅接收前端请求参数,我们可以通过@RequestParm注解去绑定请求中的参数,将(查询参数或者form表单数据)绑定到controller的方法参数中,本文结合示例代码给大家讲解的非常详细,需要的朋友可以参考下
    2023-06-06
  • java自己手动控制kafka的offset操作

    java自己手动控制kafka的offset操作

    这篇文章主要介绍了java自己手动控制kafka的offset操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2021-02-02
  • Java手动创建线程池代码实例

    Java手动创建线程池代码实例

    这篇文章主要介绍了Java手动创建线程池代码实例,FixedThreadPool或者SingleThreadPool,允许的请求队列长度为Integer.MAX_VALUE,可能会堆积大量的请求,从而导致OOM,需要的朋友可以参考下
    2023-12-12
  • Spring Boot 2.7.6整合redis与低版本的区别

    Spring Boot 2.7.6整合redis与低版本的区别

    这篇文章主要介绍了Spring Boot 2.7.6整合redis与低版本的区别,文中补充介绍了SpringBoot各个版本使用Redis之间的区别实例讲解,需要的朋友可以参考下
    2023-02-02

最新评论