Jedis的Publish/Subscribe功能的使用

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