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

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

(0)

相关推荐

  • 浅谈java实现redis的发布订阅(简单易懂)

    redis的应用场景实在太多了,现在介绍一下它的几大特性之一   发布订阅(pub/sub). 特性介绍: 什么是redis的发布订阅(pub/sub)?   Pub/Sub功能(means Publish, Subscribe)即发布及订阅功能.基于事件的系统中,Pub/Sub是目前广泛使用的通信模型,它采用事件作为基本的通信机制,提供大规模系统所要求的松散耦合的交互模式:订阅者(如客户端)以事件订阅的方式表达出它有兴趣接收的一个事件或一类事件:发布者(如服务器)可将订阅者感兴趣的事件随时通知

  • redis发布订阅_动力节点Java学院整理

    其实在很多的MQ产品中都存在这样的一个模式,我们常听到的一个例子就是邮件订阅的场景,什么意思呢,也就是说100个人订阅了你的博客,如果博主发表了文章,那么100个人就会同时收到通知邮件,除了这个场景还能找到其他场景么,当然有啦,你想想,如果你要在内存里面做一个读写分离的程序,为了维持数据的完整性,你是不是需要保证在写入的时候,也要分发到各个读内存的程序中呢?所以说场景还是很多的,在于你的挖掘~~~ 下面还是从基本命令入手: 一:命令简介 从redis手册上面可以看到,其实"发布.订阅"

  • Python redis操作实例分析【连接、管道、发布和订阅等】

    本文实例讲述了Python redis操作.分享给大家供大家参考,具体如下: 一.redis redis是一个key-value存储系统.和Memcached类似,它支持存储的value类型相对更多,包括string(字符串).list(链表).set(集合).zset(sorted set --有序集合)和hash(哈希类型).这些数据类型都支持push/pop.add/remove及取交集并集和差集及更丰富的操作,而且这些操作都是原子性的.在此基础上,redis支持各种不同方式的排序.与me

  • PHP基于Redis消息队列实现发布微博的方法

    本文实例讲述了PHP基于Redis消息队列实现发布微博的方法.分享给大家供大家参考,具体如下: phpRedisAdmin :github地址  图形化管理界面 git clone [url]https://github.com/ErikDubbelboer/phpRedisAdmin.git[/url] cd phpRedisAdmin git clone [url]https://github.com/nrk/predis.git[/url] vendor 首先安装上述的Redis图形化管理

  • redis发布和订阅_动力节点Java学院整理

    Redis 的 pub sub实现了邮件系统,发送者(在 Redis 术语中被称为发布者)发送的邮件,而接收器(用户)接收它们.由该消息传送的链路被称为信道. Redis客户端可以订阅任何数目的通道. 例子 以下举例说明如何发布用户的概念工作.在下面的例子给出一个客户端订阅一个通道名为redisChat redis 127.0.0.1:6379> SUBSCRIBE redisChat Reading messages... (press Ctrl-C to quit) 1) "subsc

  • .net core如何使用Redis发布订阅

    Redis是一个性能非常强劲的内存数据库,它一般是作为缓存来使用,但是他不仅仅可以用来作为缓存,比如著名的分布式框架dubbo就可以用Redis来做服务注册中心.接下来介绍一下.net core 使用Redis的发布/订阅功能. Redis 发布订阅 Redis 发布订阅(pub/sub)是一种消息通信模式:发送者(pub)发送消息,订阅者(sub)接收消息. Redis 客户端可以订阅任意数量的通道. 下图展示了频道 channel1 , 以及订阅这个频道的三个客户端 -- client2 .

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

    前言 Redis除了可以用作缓存数据外,另一个重要用途是它实现了发布订阅(pub/sub)消息通信模式:发送者(pub)发送消息,订阅者(sub)接收消息. 为了实现redis的发布订阅机制,首先要打开redis服务:其次,引入redis需要的jar包,在pom.xml配置文件加入以下代码: <dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> &

  • nodejs redis 发布订阅机制封装实现方法及实例代码

     nodejs redis 发布订阅机制封装 最近项目使用redis,对publish 和 subscribe的使用进行了了解,并进行了封装. var config = require('../config/config'); var log = require("./loghelp"); var redis = require("redis"); function initialclient(param) { var option={ host: config.r

  • Docker Compose+Nestjs构建Dapr Redis发布订阅分布式应用

    目录 Dapr(分布式应用程序运行时)介绍 实战 Dapr 的 Redis 发布/订阅应用 1. 创建项目 2. 创建 Dapr Placement 服务 3. 创建 Redis Publish 服务 4. 创建 Dapr Pub-Sub 组件 5. 创建 Redis Dapr Sidecar 6. 创建 NestJS Server 7. 为 NestJS 订阅服务器创建 Dockerfile 8. 将 NestJS 订阅服务添加到 docker-compose 文件 9. 创建 Dapr 订阅

  • springboot基于Redis发布订阅集群下WebSocket的解决方案

    一.背景 单机节点下,WebSocket连接成功后,可以直接发送消息.而多节点下,连接时通过nginx会代理到不同节点. 假设一开始用户连接了node1的socket服务.触发消息发送的条件的时候也通过nginx进行代理,假如代理转到了node2节点上,那么node2节点的socket服务就发送不了消息,因为一开始用户注册的是node1节点.这就导致了消息发送失败. 为了解决这一方案,消息发送时,就需要一个中间件来记录,这样,三个节点都可以获取消息,然后在根据条件进行消息推送. 二.解决方案(s

  • SpringBoot Redis 发布订阅模式(Pub/Sub)的具体使用

    目录 Redis命令行下使用发布订阅 publish 发布 subscribe 订阅 SpringBoot中使用Redis的发布订阅功能 发布者 订阅者 消息监听容器 注意:redis的发布订阅模式不可以将消息进行持久化,订阅者发生网络断开.宕机等可能导致错过消息. Redis命令行下使用发布订阅 publish 发布 发布者通过以下命令可以往指定channel发布message redis> publish channel message subscribe 订阅 订阅者通过以下命令可以订阅一

  • Redis发布订阅和实现.NET客户端详解

    前言 发布订阅在设计模式中也可以说是观察者模式,针对这个模式是处理对象间一对多的依赖关系的,当一个对象发生变化,其它依赖他的对象都要得到通知并更新. 然而它也有自己的缺点,就是当主题发生一系列的变化时,观察者都要做批量的更新,如果这样的更新成本很高,那么解决方法就是根据种类需求通知,而不能盲目的通知所有的观察者. 那针对这个缺点,一般的情况下,你没有需求谁订阅一个跟自己无关的消息推送呢?这也正好说明推送的消息需要整理而不能一窝蜂的什么消息都往一个通道里面抛,要分而治之,合理的设计发布通道的用途,

  • Android中执行java命令的方法及java代码执行并解析shell命令

    android中执行java命令的方法大家都晓得吗,下面一段内容给大家带来了具体解析. android的程序基于java开发,当我们接上调试器,执行adb shell,就可以执行linux命令,但是却并不能执行java命令. 那么在android的shell中是否就不能执行java程序了呢. 答案是否定的.我们可以通过app_process来执行java程序. 写一个hello world吧,就是刚开始学java的时候 写得那个hello world,这次要在android上运行. 用记事本新建

  • js简单粗暴的发布订阅示例代码

    什么是发布/订阅? 我打个比方,你去某个门店买衣服,你和门店店长相互并不认识,门店店长只管卖他的衣服,并不关心是谁来买,而你也只管买你想要的衣服,并不关心是哪个门店在卖,这时,门店和你就组成了一个发布/订阅的关系. 当门店挂出衣服款式,你去找你想要的衣服,如果找到了,那就买下来,如果没找到,那就离开这家店.整个过程就是这么简单. 使用场景 异步通信.多页面间相互通信,pageA 的方法想在 pageB的方法调用的某个时机触发 直接上代码 class Publish { constructor()

随机推荐