redis釋出訂閱Java程式碼實現

摘要： Redis除了可以用作快取資料外，另一個重要用途是它實現了釋出訂閱(pub/sub)訊息通訊模式：傳送者(pub)傳送訊息，訂閱者(sub)接收訊息。 為了實現redis的釋出訂閱機制，首先要開啟redis服務；其次，引入redis需要的jar包，在pom.xml配置檔案加入以下程式碼： ...

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