閱讀212 返回首頁    go 魔獸

消息發布與訂閱__最佳實踐_雲數據庫 Redis 版-阿裏雲

消息的發布與訂閱

場景介紹

ApsaraDB for Redis 也提供了與 Redis 相同的消息發布（pub）與訂閱（sub）功能。即一個 client 發布消息，其他多個 client 訂閱消息。

需要注意的是，ApsaraDB for Redis 發布的消息是“非持久”的，即消息發布者隻負責發送消息，而不管消息是否有接收方，也不會保存之前發送的消息，即發布的消息“即發即失”；消息訂閱者也隻能得到訂閱之後的消息，頻道（channel）中此前的消息將無從獲得。

此外，消息發布者（即 publish 客戶端）無需獨占與服務器端的連接，您可以在發布消息的同時，使用同一個客戶端連接進行其他操作（例如 List 操作等）。但是，消息訂閱者（即 subscribe 客戶端）需要獨占與服務器端的連接，即進行 subscribe 期間，該客戶端無法執行其他操作，而是以阻塞的方式等待頻道（channel）中的消息；因此消息訂閱者需要使用單獨的服務器連接，或者需要在單獨的線程中使用（參見如下示例）。

代碼示例

消息發布者 (即 publish client)

  1. package message.kvstore.aliyun.com;
  3. import redis.clients.jedis.Jedis;
  5. public class KVStorePubClient {
  7.     private Jedis jedis;//
  8.     public KVStorePubClient(String host,int port, String password){
  9.         jedis = new Jedis(host,port);
  11.         //KVStore的實例ID及密碼
  12.         String authString = jedis.auth(password);//password
  14.         if (!authString.equals("OK"))
  15.         {
  16.             System.err.println("AUTH Failed: " + authString);
  17.             return;
  18.         }
  19.     }
  21.     public void pub(String channel,String message){
  22.         System.out.println("  >>> 發布(PUBLISH) > Channel:"+channel+" > 發送出的Message:"+message);
  24.         jedis.publish(channel, message);
  25.     }
  27.     public void close(String channel){
  28.         System.out.println("  >>> 發布(PUBLISH)結束 > Channel:"+channel+" > Message:quit");
  30.         //消息發布者結束發送，即發送一個“quit”消息；
  31.         jedis.publish(channel, "quit");
  33.     }
  35. }

消息訂閱者 (即 subscribe client)

  1. package message.kvstore.aliyun.com;
  3. import redis.clients.jedis.Jedis;
  4. import redis.clients.jedis.JedisPubSub;
  6. public class KVStoreSubClient extends Thread{
  8.     private Jedis jedis;
  9.     private String channel;
  10.     private JedisPubSub listener;
  12.     public KVStoreSubClient(String host,int port, String password){
  13.         jedis = new Jedis(host,port);
  15.         //ApsaraDB for Redis的實例ID及密碼
  16.                 String authString = jedis.auth(password);//kvstore_instance_id:password
  18.                 if (!authString.equals("OK"))
  19.                 {
  20.                     System.err.println("AUTH Failed: " + authString);
  21.                     return;
  22.                 }
  23.     }
  25.     public void setChannelAndListener(JedisPubSub listener,String channel){
  26.         this.listener=listener;
  27.         this.channel=channel;
  28.     }
  30.     private void subscribe(){
  32.         if(listener==null || channel==null){
  33.             System.err.println("Error:SubClient> listener or channel is null");
  34.         }
  36.         System.out.println("  >>> 訂閱(SUBSCRIBE) > Channel:"+channel);
  37.         System.out.println();
  39.         //接收者在偵聽訂閱的消息時，將會阻塞進程，直至接收到quit消息（被動方式），或主動取消訂閱
  40.         jedis.subscribe(listener, channel);
  41.     }
  43.     public void unsubscribe(String channel){
  44.         System.out.println("  >>> 取消訂閱(UNSUBSCRIBE) > Channel:"+channel);
  45.         System.out.println();
  47.         listener.unsubscribe(channel);
  48.     }
  50.     @Override
  51.     public void run() {
  52.         try{
  54.             System.out.println();
  55.             System.out.println("----------訂閱消息SUBSCRIBE 開始-------");
  57.             subscribe();
  59.             System.out.println("----------訂閱消息SUBSCRIBE 結束-------");
  60.             System.out.println();
  61.         }catch(Exception e){
  62.             e.printStackTrace();
  63.         }
  65.     }
  67. }

消息監聽者

  1. package message.kvstore.aliyun.com;
  3. import redis.clients.jedis.JedisPubSub;
  5. public class KVStoreMessageListener extends JedisPubSub{
  7.     @Override
  8.     public void onMessage(String channel, String message) {
  10.         System.out.println("  <<< 訂閱(SUBSCRIBE)< Channel:" + channel + " >接收到的Message:" + message );
  11.         System.out.println();
  13.         //當接收到的message為quit時，取消訂閱(被動方式)
  14.         if(message.equalsIgnoreCase("quit")){
  15.             this.unsubscribe(channel);
  16.         }
  17.     }
  19.     @Override
  20.     public void onPMessage(String pattern, String channel, String message) {
  21.         // TODO Auto-generated method stub
  23.     }
  25.     @Override
  26.     public void onSubscribe(String channel, int subscribedChannels) {
  27.         // TODO Auto-generated method stub
  29.     }
  31.     @Override
  32.     public void onUnsubscribe(String channel, int subscribedChannels) {
  33.         // TODO Auto-generated method stub
  35.     }
  37.     @Override
  38.     public void onPUnsubscribe(String pattern, int subscribedChannels) {
  39.         // TODO Auto-generated method stub
  41.     }
  43.     @Override
  44.     public void onPSubscribe(String pattern, int subscribedChannels) {
  45.         // TODO Auto-generated method stub
  47.     }
  48. }

示例主程序

  1. package message.kvstore.aliyun.com;
  3. import java.util.UUID;
  5. import redis.clients.jedis.JedisPubSub;
  7. public class KVStorePubSubTest {
  9.     //ApsaraDB for Redis的連接信息，從控製台可以獲得
  10.     static final String host = "xxxxxxxxxx.m.cnhza.kvstore.aliyuncs.com";
  11.     static final int port = 6379;
  12.     static final String password="xxxxxxxxxx:yyyyyyyy";//kvstore_instance_id:password
  14.     public static void main(String[] args) throws Exception{
  15.             KVStorePubClient pubClient = new KVStorePubClient(host, port,password);
  17.             final String channel = "KVStore頻道-A";
  19.             //消息發送者開始發消息，此時還無人訂閱，所以此消息不會被接收
  20.             pubClient.pub(channel, "Aliyun消息1：（此時還無人訂閱，所以此消息不會被接收）");
  22.             //消息接收者
  23.             KVStoreSubClient subClient = new KVStoreSubClient(host, port,password);
  25.             JedisPubSub listener = new KVStoreMessageListener();
  26.             subClient.setChannelAndListener(listener, channel);
  28.             //消息接收者開始訂閱
  29.             subClient.start();
  32.             //消息發送者繼續發消息
  33.             for (int i = 0; i < 5; i++) {
  35.                 String message=UUID.randomUUID().toString();
  36.                 pubClient.pub(channel, message);
  37.                 Thread.sleep(1000);
  38.             }
  40.             //消息接收者主動取消訂閱
  41.             subClient.unsubscribe(channel);
  43.             Thread.sleep(1000);
  44.             pubClient.pub(channel, "Aliyun消息2：（此時訂閱取消，所以此消息不會被接收）");
  46.             //消息發布者結束發送，即發送一個“quit”消息；
  47.             //此時如果有其他的消息接收者，那麼在listener.onMessage()中接收到“quit”時，將執行“unsubscribe”操作。
  48.             pubClient.close(channel);
  50.         }
  52.     }

運行結果

在輸入了正確的 ApsaraDB for Redis 實例訪問地址和密碼之後，運行以上 Java 程序，輸出結果如下。

  1.   >>> 發布(PUBLISH) > Channel:KVStore頻道-A > 發送出的Message:Aliyun消息1：（此時還無人訂閱，所以此消息不會被接收）
  3. ----------訂閱消息SUBSCRIBE 開始-------
  4.   >>> 訂閱(SUBSCRIBE) > Channel:KVStore頻道-A
  6.   >>> 發布(PUBLISH) > Channel:KVStore頻道-A > 發送出的Message:0f9c2cee-77c7-4498-89a0-1dc5a2f65889
  7.   <<< 訂閱(SUBSCRIBE)< Channel:KVStore頻道-A >接收到的Message:0f9c2cee-77c7-4498-89a0-1dc5a2f65889
  9.   >>> 發布(PUBLISH) > Channel:KVStore頻道-A > 發送出的Message:ed5924a9-016b-469b-8203-7db63d06f812
  10.   <<< 訂閱(SUBSCRIBE)< Channel:KVStore頻道-A >接收到的Message:ed5924a9-016b-469b-8203-7db63d06f812
  12.   >>> 發布(PUBLISH) > Channel:KVStore頻道-A > 發送出的Message:f1f84e0f-8f35-4362-9567-25716b1531cd
  13.   <<< 訂閱(SUBSCRIBE)< Channel:KVStore頻道-A >接收到的Message:f1f84e0f-8f35-4362-9567-25716b1531cd
  15.   >>> 發布(PUBLISH) > Channel:KVStore頻道-A > 發送出的Message:746bde54-af8f-44d7-8a49-37d1a245d21b
  16.   <<< 訂閱(SUBSCRIBE)< Channel:KVStore頻道-A >接收到的Message:746bde54-af8f-44d7-8a49-37d1a245d21b
  18.   >>> 發布(PUBLISH) > Channel:KVStore頻道-A > 發送出的Message:8ac3b2b8-9906-4f61-8cad-84fc1f15a3ef
  19.   <<< 訂閱(SUBSCRIBE)< Channel:KVStore頻道-A >接收到的Message:8ac3b2b8-9906-4f61-8cad-84fc1f15a3ef
  21.   >>> 取消訂閱(UNSUBSCRIBE) > Channel:KVStore頻道-A
  23. ----------訂閱消息SUBSCRIBE 結束-------
  25.   >>> 發布(PUBLISH) > Channel:KVStore頻道-A > 發送出的Message:Aliyun消息2：（此時訂閱取消，所以此消息不會被接收）
  26.   >>> 發布(PUBLISH)結束 > Channel:KVStore頻道-A > Message:quit

以上示例中僅演示了一個發布者與一個訂閱者的情況，實際上發布者與訂閱者都可以為多個，發送消息的頻道（channel）也可以是多個，對以上代碼稍作修改即可。

最後更新：2016-12-16 17:21:38

  上一篇：go 商品相關性分析__最佳實踐_雲數據庫 Redis 版-阿裏雲
  下一篇：go 管道傳輸__最佳實踐_雲數據庫 Redis 版-阿裏雲