使用Spring Data Redis操作Redis（二）

上一篇講述了Spring Date Redis操作Redis的大部分主題 ,本篇介紹Redis的訂閱和發布功能在Spring應用中的使用。

1.Redis的Pub/Sub命令

Redis的訂閱和發布服務有如下圖6個命令，下面分別對每個命令做簡單說明。

publish: 向指定的channel(頻道)發送message(消息)

subscribe:訂閱指定channel，可以一次訂閱多個

psubscribe:訂閱指定pattern(模式，具有頻道名的模式匹配)的頻道

unsubscribe:取消訂閱channel,可以一次取消多個訂閱

punsubscribe:取消指定pattern的訂閱

pubsub:是一個查看訂閱與發布系統狀態的內省命令,它由數個不同格式的子命令組成(具體可參見：

http://redis.io/commands/pubsub)

在SDR（Spring Data Redis）中頻道對應Topic類，Top類是一個介面有Channel和Pattern兩個實現類，分別是指定名稱的頻道和模式匹配的頻道。關於訂閱信息由Subscription介面定義。

2.Redis消息監聽容器聲明和消息監聽器註冊

在SDR中可以用兩種方式來實現消息監聽容器的聲明，一種是通過Redis的命名空間，一種是定義Bean。

這裡主要涉及到RedisMessageListenerContainer，MessageListenerAdapter，MessageListener幾個類。

2.1使用Redis命名空間的方式配置

<bean id="mdpListener" class="secondriver.spring.redis.MyMessageListener" /> <bean id="mdelegateListener" class="secondriver.spring.redis.DefaultMessageDelegate" /> <redis:listener-container connection-factory="jedisConnectionFactory"> <redis:listener ref="mdpListener" topic="spring*" /> <redis:listener ref="mdelegateListener" method="handleMessage" topic="cctv5 cctv6 nbtv hello*" /> </redis:listener-container>

說明:

定義topic可以是具體的channel的名字也可以是Pattern，多個頻道（主題）用空格隔開即可。

這裡定義了兩個Listner，MyMessageListener實現了MessageaListener介面,DefaultMessageDelegate實現了MessageDelegate介面。

MyMessageListener:

package secondriver.spring.redis;import org.springframework.data.redis.connection.Message;import org.springframework.data.redis.connection.MessageListener;public class MyMessageListener implements MessageListener { @Override public void onMessage(Message message, byte[] pattern) { System.out.println("channel:" + new String(message.getChannel()) + ",message:" + new String(message.getBody())); }}

MessageDelegate介面：

package secondriver.spring.redis;import java.io.Serializable;import java.util.Map;public interface MessageDelegate { public void handleMessage(String message); public void handleMessage(Map<?, ?> message); public void handleMessage(byte[] message); public void handleMessage(Serializable message); // pass the channel/pattern as well public void handleMessage(Serializable message, String channel);}

DefaultMessageDelegate類型：

package secondriver.spring.redis;import java.io.Serializable;import java.util.Map;public class DefaultMessageDelegate implements MessageDelegate { @Override public void handleMessage(String message) { System.out.println("handleMessage(String message):" + message); } @Override public void handleMessage(Map<?, ?> message) { System.out.println("handleMessage(Map<?, ?> message):" + message); } @Override public void handleMessage(byte[] message) { System.out.println("handleMessage(byte[] message):" + new String(message)); } @Override public void handleMessage(Serializable message) { System.out.println("handleMessage(Serializable message):" + message.toString()); } @Override public void handleMessage(Serializable message, String channel) { System.out .println("handleMessage(Serializable message, String channel):" + message.toString() + ", channel:" + channel); }}

這種定義消息監聽的方式不依賴於Redis，其被設計為一個message-driven POJOs (MDPs)，MessageListenerAdapter實現了MessageListener介面，它將會把Message委託給目標監聽器（Target Listener）DefalutMessageDelegate的方法，對Message參數的適當轉換，然後通過反射來調用方法。

2.2定義Bean的方式配置

<!-- Bean Configuration --> <bean id="messageListener" class="org.springframework.data.redis.listener.adapter.MessageListenerAdapter"> <constructor-arg> <bean class="secondriver.spring.redis.MyMessageListener" /> </constructor-arg> </bean> <bean id="redisContainer" class="org.springframework.data.redis.listener.RedisMessageListenerContainer"> <property name="connectionFactory" ref="jedisConnectionFactory" /> <property name="messageListeners"> <map> <entry key-ref="messageListener"> <list> <bean class="org.springframework.data.redis.listener.ChannelTopic"> <constructor-arg value="springtv" /> </bean> <bean class="org.springframework.data.redis.listener.PatternTopic"> <constructor-arg value="hello*" /> </bean> <bean class="org.springframework.data.redis.listener.PatternTopic"> <constructor-arg value="tv*" /> </bean> </list> </entry> </map> </property> </bean>

3.模擬消息的發布和接收

// 簡單測試RedisMessageListener @Ignore @Test public void test10() throws InterruptedException { RedisMessageListenerContainer rmlc; // ctx.getBean(RedisMessageListenerContainer.class); rmlc = (RedisMessageListenerContainer) ctx.getBean("redisContainer"); while (true) { if (rmlc.isRunning()) { System.out .println("RedisMessageListenerContainer is running.."); } Thread.sleep(5000); } }

由於這裡是測試，通過一個死循環來保持程序一直運行，然後向Redis服務的指定頻道發布消息，則符合訂閱的頻道的消息將被客戶端連接接收到，並且MyMessageListener對象中的onMessage方法被調用。

下圖是模擬過程：

發布了四條消息，分別是spring,springtv，hello， nono四個頻道，根據定義bean這種配置中的Topic名稱，只有springtv,hello符合模式匹配，並且也同樣收到了這兩條消息。

除了這種模擬外，實際在應用開發中是通過JedisConnection的Pub/Sub相關的方法來向Redis服務發布消息的或者RedisTemplate的convertAndSend方法。

4.最後

SDR正處於發展階段的項目，更多特性閱讀源代碼，一步步挖掘。