開啟redis伺服器

1. [[email protected] ~]# redis-server /opt/redis-2.4.10/redis.conf  
2. [7719] 16 Apr 11:37:22 # Warning: 32 bit instance detected but no memory limit set. Setting 3.5 GB maxmemory limit with 'noeviction' policy now. 
3. [7719] 16 Apr 11:37:22 * Server started, Redis version 2.4.10 
4. [7719] 16 Apr 11:37:22 # WARNING overcommit_memory is set to 0! Background save may fail under low memory condition. To fix this issue add 'vm.overcommit_memory
    = 1' to /etc/sysctl.conf and then reboot or run the command 'sysctl vm.overcommit_memory=1' for this to take effect. 

 開啟一個客戶端訂閱一個news.sports的channel。

1. [[email protected] ~]# redis-cli 
2. redis 127.0.0.1:6379> subscribe news.sports 
3. Reading messages... (press Ctrl-C to quit) 
4. 1) "subscribe" 
5. 2) "news.sports" 
6. 3) (integer) 1 

 可以看到已經開始了監聽，向news.sports channel釋出一條訊息

1. [[email protected] ~]# redis-cli             
2. redis 127.0.0.1:6379> publish news.sports "kaka is back" 
3. (integer) 1 

 訂閱的客戶端順利收到訊息

1. redis 127.0.0.1:6379> subscribe news.sports 
2. Reading messages... (press Ctrl-C to quit) 
3. 1) "subscribe" 
4. 2) "news.sports" 
5. 3) (integer) 1 
6. 1) "message" 
7. 2) "news.sports" 
8. 3) "kaka is back" 

 接下來使用jedis來進行釋出/訂閱的驗證

 釋出訊息是通過jedis.publish(String channel, String message)來發布的，其實就是往redis伺服器釋出一條publish命令。

1. public void publish(final byte[] channel, final byte[] message) { 
2.        sendCommand(PUBLISH, channel, message); 
3.    } 

 訂閱訊息是通過jedis.subscribe（JedisPub pub,String channel）來進行的，channel好理解，那麼JedisPub是什麼呢。

 看原始碼吧。

 Jedis訂閱方法的原始碼為

1. public void subscribe(JedisPubSub jedisPubSub, String... channels) { 
2.       checkIsInMulti(); 
3.       connect(); 
4.       client.setTimeoutInfinite(); 
5.       jedisPubSub.proceed(client, channels); 
6.       client.rollbackTimeout(); 
7.   } 

可以看到，主要是通過jedisPubSub.proceed(client, channels);來進行訂閱的。看proceed方法。

1. public void proceed(Client client, String... channels) { 
2.        this.client = client; 
3.        client.subscribe(channels); 
4.        client.flush(); 
5.        process(client); 
6.    } 

追蹤client.subscribe(channels)可以看到，

1. public void subscribe(final byte[]... channels) { 
2.        sendCommand(SUBSCRIBE, channels); 
3.    } 

其只是向伺服器傳送了一個subcribe的命令而已。

那麼要了解jedisPubSub的作用，只能看process方法了。簡單看process其實是一個do...while迴圈

1. private void process(Client client) { 
2.        do { 
3.        } while (isSubscribed()); 
4.    } 

我們可以猜測正是靠著這個迴圈來不斷的讀取伺服器那邊傳到來的訂閱的訊息。

看主體

1. List<Object>reply = client.getObjectMultiBulkReply(); 
2.            final Object firstObj = reply.get(0); 
3.            if (!(firstObj instanceof byte[])) { 
4.                throw new JedisException("Unknown message type: " + firstObj); 
5.            } 
6.            final byte[] resp = (byte[]) firstObj; 
7.            if (Arrays.equals(SUBSCRIBE.raw, resp)) { 
8.                subscribedChannels = ((Long) reply.get(2)).intValue(); 
9.                final byte[] bchannel = (byte[]) reply.get(1); 
10.                final String strchannel = (bchannel == null) ? null 
11.                        : SafeEncoder.encode(bchannel); 
12.                onSubscribe(strchannel, subscribedChannels); 
13.            } else if (Arrays.equals(UNSUBSCRIBE.raw, resp)) { 
14.                subscribedChannels = ((Long) reply.get(2)).intValue(); 
15.                final byte[] bchannel = (byte[]) reply.get(1); 
16.                final String strchannel = (bchannel == null) ? null 
17.                        : SafeEncoder.encode(bchannel); 
18.                onUnsubscribe(strchannel, subscribedChannels); 
19.            } else if (Arrays.equals(MESSAGE.raw, resp)) { 
20.                final byte[] bchannel = (byte[]) reply.get(1); 
21.                final byte[] bmesg = (byte[]) reply.get(2); 
22.                final String strchannel = (bchannel == null) ? null 
23.                        : SafeEncoder.encode(bchannel); 
24.                final String strmesg = (bmesg == null) ? null : SafeEncoder 
25.                        .encode(bmesg); 
26.                onMessage(strchannel, strmesg); 
27.            } else if (Arrays.equals(PMESSAGE.raw, resp)) { 
28.                final byte[] bpattern = (byte[]) reply.get(1); 
29.                final byte[] bchannel = (byte[]) reply.get(2); 
30.                final byte[] bmesg = (byte[]) reply.get(3); 
31.                final String strpattern = (bpattern == null) ? null 
32.                        : SafeEncoder.encode(bpattern); 
33.                final String strchannel = (bchannel == null) ? null 
34.                        : SafeEncoder.encode(bchannel); 
35.                final String strmesg = (bmesg == null) ? null : SafeEncoder 
36.                        .encode(bmesg); 
37.                onPMessage(strpattern, strchannel, strmesg); 
38.            } else if (Arrays.equals(PSUBSCRIBE.raw, resp)) { 
39.                subscribedChannels = ((Long) reply.get(2)).intValue(); 
40.                final byte[] bpattern = (byte[]) reply.get(1); 
41.                final String strpattern = (bpattern == null) ? null 
42.                        : SafeEncoder.encode(bpattern); 
43.                onPSubscribe(strpattern, subscribedChannels); 
44.            } else if (Arrays.equals(PUNSUBSCRIBE.raw, resp)) { 
45.                subscribedChannels = ((Long) reply.get(2)).intValue(); 
46.                final byte[] bpattern = (byte[]) reply.get(1); 
47.                final String strpattern = (bpattern == null) ? null 
48.                        : SafeEncoder.encode(bpattern); 
49.                onPUnsubscribe(strpattern, subscribedChannels); 
50.            } else { 
51.                throw new JedisException("Unknown message type: " + firstObj); 
52.            } 

可以看到，通過client.getObjectMultiBulkReply()來得到返回來的訊息。判斷訊息的型別來進行不同的操作。比如Arrays.equals(SUBSCRIBE.raw, resp)判斷返回來的訊息是訂閱，subscribedChannels = ((Long) reply.get(2)).intValue();是取得訊息，也是do...while判斷迴圈的條件，也就是說這一次如果讀到訊息了，則進行下一次迴圈。那麼onSubscribe(String channel, int subscribedChannels)究竟做了什麼事，看開頭

1. public abstract void onMessage(String channel, String message); 
2.    public abstract void onPMessage(String pattern, String channel, 
3.            String message); 
4.    public abstract void onSubscribe(String channel, int subscribedChannels); 
5.    public abstract void onUnsubscribe(String channel, int subscribedChannels); 
6.    public abstract void onPUnsubscribe(String pattern, int subscribedChannels); 
7.    public abstract void onPSubscribe(String pattern, int subscribedChannels); 

可以看到這是xetorthio留給我們的方法。onSubscrible是訂閱時應該做些什麼，onMessage就是有訊息傳來是做些什麼，以此類推。

接下來可以寫一個方法來發布和訂閱訊息了。

1. package redis.client.jredis.tests; 
2. import java.util.Timer; 
3. import java.util.TimerTask; 
4. import org.junit.Test; 
5. import redis.clients.jedis.Jedis; 
6. import redis.clients.jedis.JedisPool; 
7. import redis.clients.jedis.JedisPoolConfig; 
8. import redis.clients.jedis.JedisPubSub; 
9. publicclass JedisTest extends JedisTestBase { 
10.     JedisPool pool = null; 
11.     /** 
12.      * 測試釋出驗證 
13.      */
14.     @Test
15.     publicvoid testPS(){ 
16.         /** 
17.         Jedis jedis = new Jedis("192.168.5.146",6379); 
18.         jedis.set("name", "xiaoruoen"); 
19.         jedis.publish("news.blog.title", "Hello,World"); 
20.         //*/
21.         final String host = "192.168.5.146"; 
22.         JedisPoolConfig config = new JedisPoolConfig(); 
23.         pool = new JedisPool(new JedisPoolConfig(),host); 
24.         subscribe(new NewsListener(), "news.sports"); 
25.         Timer timer = new Timer(); 
26.         timer.schedule(new TimerTask() { 
27.             @Override
28.             publicvoid run() { 
29.                 // TODO Auto-generated method stub
30.                 publish("news.sports", "{\"_id\":335566,\"author\":\"xiaoruoen\",\"title\":\"kaka is back\"}"); 
31.             } 
32.         }, 1000, 3000); 
33.     } 
34.     public Jedis getResource(int dbnum){ 
35.         Jedis jedis = pool.getResource(); 
36.         jedis.select(dbnum); 
37.         return jedis; 
38.     } 
39.     /** 
40.      *  
41.      * @param channel 
42.      * @param message ex:"{\"_id\":335566,\"author\":\"xiaoruoen\",\"title\":\"kaka is back\"}" 
43.      */
44.     publicvoid publish(String channel,String message){ 
45.         Jedis jedis = getResource(12); 
46.         jedis.publish(channel, message); 
47.         pool.returnResource(jedis); 
48.     } 
49.     publicvoid subscribe(JedisPubSub listener,String channel){ 
50.         Jedis jedis = getResource(12); 
51.         jedis.subscribe(listener, channel); 
52.         pool.returnResource(jedis); 
53.     } 
54. } 

1. package redis.client.jredis.tests; 
2. import redis.clients.jedis.JedisPubSub; 
3. publicclass NewsListener extends JedisPubSub { 
4.     @Override
5.     publicvoid onMessage(String channel, String message) { 
6.         System.out.println("get message from"+channel+"   "+message); 
7.     } 
8.     @Override
9.     publicvoid onPMessage(String pattern, String channel, String message) { 
10.         System.out.println("get message from"+channel+"   "+message); 
11.     } 
12.     @Override
13.     publicvoid onSubscribe(String channel, int subscribedChannels) { 
14.         System.out.println("subscribe the channel:"+channel); 
15.     } 
16.     @Override
17.     publicvoid onUnsubscribe(String channel, int subscribedChannels) { 
18.         System.out.println("get message from"+channel); 
19.     } 
20.     @Override
21.     publicvoid onPUnsubscribe(String pattern, int subscribedChannels) { 
22.         System.out.println("get message from"+subscribedChannels); 
23.     } 
24.     @Override
25.     publicvoid onPSubscribe(String pattern, int subscribedChannels) { 
26.         System.out.println("get message from"+subscribedChannels); 
27.     } 
28. } 

發現只打印了一條資料subscribe the channel:news.sports

沒按我們所期望的那樣那所有釋出的訊息都打印出來。

到官網檢視

看到Note that subscribe is a blocking operation operation because it will poll Redis for responses on the thread that calls subscribe.可以看到subcribe是一個執行緒中的塊操作。我猜測是在釋出與接收的過程中，如果在同一執行緒裡面進行操作，一邊阻塞著流，另一邊無法進行操作。於是將publish改寫為另一執行緒啟動。修改如下：

1. publicstaticvoid main(String[] args){ 
2.         final String host = "192.168.5.146"; 
3.         JedisPoolConfig config = new JedisPoolConfig(); 
4.         pool = new JedisPool(new JedisPoolConfig(),host); 
5.         Thread thread = new Thread(new Test().new PublishThread()); 
6.         thread.start(); 
7.         subscribe(new NewsListener(), "news.sports"); 
8. } 

1. class PublishThread implements Runnable{ 
2.         @Override 
3.         public void run() { 
4.             Timer timer = new Timer(); 
5.             timer.schedule(new TimerTask() { 
6.                 @Override 
7.                 public void run() { 
8.                     // TODO Auto-generated method stub 
9.                     publish("news.sports", "{\"_id\":335566,\"author\":\"xiaoruoen\",\"title\":\"kaka is back\"}"); 
10.                 } 
11.             }, 1000, 3000); 
12.         }  
13.     } 

最終釋出訂閱成功。

1. Pool:[email protected] 
2. subscribe the channel:news.sports 
3. Pool:[email protected] 
4. get message fromnews.sports   {"_id":335566,"author":"xiaoruoen","title":"kaka is back"} 
5. Pool:[email protected] 
6. get message fromnews.sports   {"_id":335566,"author":"xiaoruoen","title":"kaka is back"} 
7. Pool:[email protected] 
8. get message fromnews.sports   {"_id":335566,"author":"xiaoruoen","title":"kaka is back"} 
9. Pool:[email protected] 
10. get message fromnews.sports   {"_id":335566,"author":"xiaoruoen","title":"kaka is back"} 
11. Pool:[email protected]