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>

1、訂閱消息（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消息。

2、發佈消息

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