redis內置了發布/訂閱功能,可以作為消息機制使用。所以這里主要使用Jedis的Publish/Subscribe功能。
1、使用Spring來配置Jedis連接池
<!-- pool配置 --> <bean id="jedisPoolConfig" class="redis.clients.jedis.JedisPoolConfig"> <property name="maxActive" value="20" /> <property name="maxIdle" value="10" /> <property name="maxWait" value="1000" /> <property name="testOnBorrow" value="true" /> </bean> <!-- jedis pool配置 --> <bean id="jedisPool" class="redis.clients.jedis.JedisPool"> <constructor-arg index="0" ref="jedisPoolConfig" /> <constructor-arg index="1" value="127.0.0.1" /> <constructor-arg index="2" value="6389"/> <constructor-arg index="3" value="30000"/> <constructor-arg index="4" value="123456789"/> </bean>
2、編寫Lister
要使用Jedis的Publish/Subscribe功能,必須編寫對JedisPubSub的自己的實現,其中的函數的功能如下:
package demo; import redis.clients.jedis.JedisPubSub; public class RedisSubPubListener extends JedisPubSub { // 取得訂閱的消息后的處理 public void onMessage(String channel, String message) { //TODO:接收訂閱頻道消息后,業務處理邏輯 System.out.println(channel + "=" + message); } // 初始化訂閱時候的處理 public void onSubscribe(String channel, int subscribedChannels) { // System.out.println(channel + "=" + subscribedChannels); } // 取消訂閱時候的處理 public void onUnsubscribe(String channel, int subscribedChannels) { // System.out.println(channel + "=" + subscribedChannels); } // 初始化按表達式的方式訂閱時候的處理 public void onPSubscribe(String pattern, int subscribedChannels) { // System.out.println(pattern + "=" + subscribedChannels); } // 取消按表達式的方式訂閱時候的處理 public void onPUnsubscribe(String pattern, int subscribedChannels) { // System.out.println(pattern + "=" + subscribedChannels); } // 取得按表達式的方式訂閱的消息后的處理 public void onPMessage(String pattern, String channel, String message) { System.out.println(pattern + "=" + channel + "=" + message); }
}
3、實現訂閱動能
Jedis有兩種訂閱模式:subsribe(一般模式設置頻道)和psubsribe(使用模式匹配來設置頻道)。不管是那種模式都可以設置個數不定的頻道。訂閱得到信息在將會lister的onMessage(…)方法或者onPMessage(…)中進行進行處理,這里我們只是做了簡單的輸出。
這里啟動了訂閱監聽,線程將在這里被阻塞,訂閱得到信息在lister的onMessage(…)方法或者onPMessage(…)方法中進行處理
@Autowired private JedisPool jedisPool; @Test public void testSub() { final RedisSubPubListener listener = new RedisSubPubListener(); Jedis jedis = jedisPool.getResource(); jedis.subscribe(listener, "channel"); }
//可以訂閱多個頻道
jedis.subscribe(listener, "foo", "bar");
//也用數組的方式設置多個頻道
jedis.subscribe(listener, new String[]{"hello_foo","hello_test"});
//使用模式匹配的方式設置頻道
jedis.psubscribe(listener, new String[]{"hello_*"});
4、實現發布端代碼
發布消息只用調用Jedis的publish(…)方法即可。
@Test public void testPub(){ Jedis jedis = jedisPool.getResource(); jedis.publish("channel", "bar123"); System.out.println("==========已經發布消息==================="); }
注:ShardedJedis沒有發布和訂閱功能,需要還原為Jedis,獲取方式如下(完全可以重新配置一份JedisPool):
ShardedJedis shardedJedis =shardedJedisPool.getResource(); Jedis[] jedisArray = new Jedis[]{}; jedisArray = shardedJedis.getAllShards().toArray(jedisArray); Jedis jedis = jedisArray[0];