美文网首页
redis发布订阅Java代码实现

redis发布订阅Java代码实现

作者: 哦00 | 来源:发表于2019-02-14 17:50 被阅读0次

    Redis除了可以用作缓存数据外,另一个重要用途是它实现了发布订阅(pub/sub)消息通信模式:发送者(pub)发送消息,订阅者(sub)接收消息。

    为了实现redis的发布订阅机制,首先要打开redis服务;其次,引入redis需要的jar包,在pom.xml配置文件加入以下代码:

    redis.clients

    jedis

    2.1.0

    由于订阅消息通道需要再tomcat启动时触发,因此,需要创建一个listener监听器,在监听器里实现redis订阅,在web.xml里配置监听器如下:

    com.test.listener.InitListener

    一、订阅消息(InitListener实现)

    redis支持多通道订阅,一个客户端可以同时订阅多个消息通道,如下代码所示,订阅了13个通道。由于订阅机制是线程阻塞的,需要额外开启一个线程专门用于处理订阅消息及接收消息处理。

    publicclassInitListenerimplements ServletContextListener{

        privateLogger logger = Logger.getLogger(InitListener.class);

        @Override

        publicvoid contextInitialized(ServletContextEvent sce) {

            logger.info("启动tomcat");// 连接redisMap proMap = PropertyReader.getProperties();

            finalString url = proMap.get("redis.host");

            finalInteger port = Integer.parseInt(proMap.get("redis.port"));finalClassPathXmlApplicationContext classPathXmlApplicationContext =newClassPathXmlApplicationContext("classpath*:applicationContext.xml");

            finalRedisSubListener redisSubListener = (RedisSubListener) classPathXmlApplicationContext.getBean("redisSubListener");

            // 为防止阻塞tomcat启动,开启线程执行newThread(new Runnable(){ 

                publicvoid run(){ 

                    // 连接redis,建立监听Jedis jedis =null;

                    while(true){

                        //解码资源更新通知,画面选看回复,画面选看停止回复,预案启动,预案停止,轮切启动,轮切停止,预案启动回复,预案停止回复,轮切启动回复,轮切停止回复,监视屏分屏状态通知,画面状态通知String[] channels =newString[] { "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配置文件里接入以下配置:

                                           

    RedisMessageServiceImpl用于处理接收的redis消息。

    二、发布消息

    publicclass RedisPublishUtil {

        privateLogger logger = Logger.getLogger(RedisPublishUtil.class);

        publicstatic Jedis pubJedis;

        privatestaticMap proMap = PropertyReader.getProperties();

        privatestaticfinalString redisPort = proMap.get("redis.port");

        privatestaticString url = proMap.get("redis.host");

        privatestaticfinalintport = Integer.parseInt(redisPort);

        publicvoid setPubJedis(Jedis jedis) {

            RedisPublishUtil.pubJedis = jedis;

        }public Jedis getPubJedis() {

            if(pubJedis ==null) {

                createJedisConnect();

            }

            // 返回对象return pubJedis;

        }

        public Jedis createJedisConnect(){

            // 连接redislogger.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;

        }    //公共发布接口

        publicvoid 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);

            }

        }

    }

    publicclass PropertyReader {

        privatestaticLogger logger = Logger.getLogger(PropertyReader.class);

        /*    * 获得数据库链接的配置文件

        */publicstaticMap getProperties(){

            logger.info("读取redis配置文件开始。。。");

            Properties prop =new Properties();   

            Map proMap  =newHashMap();

            try {

                //读取属性文件redis.propertiesInputStream in= PropertyReader.class.getClassLoader().getResourceAsStream("redis.properties"); 

                prop.load(in);    ///加载属性列表Iterator 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工程师朋友们加入Java群: 741514154

    群内提供免费的Java架构学习资料(里面有高可用、高并发、高性能及分布式、Jvm性能调优、Spring源码,MyBatis,Netty,Redis,Kafka,Mysql,Zookeeper,Tomcat,Docker,Dubbo,Nginx等多个知识点的架构资料)合理利用自己每一分每一秒的时间来学习提升自己,不要再用"没有时间“来掩饰自己思想上的懒惰!趁年轻,使劲拼,给未来的自己一个交代!

    相关文章

      网友评论

          本文标题:redis发布订阅Java代码实现

          本文链接:https://www.haomeiwen.com/subject/pthyeqtx.html