阅读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;
  2. import redis.clients.jedis.Jedis;
  3. public class KVStorePubClient {
  4. private Jedis jedis;//
  5. public KVStorePubClient(String host,int port, String password){
  6. jedis = new Jedis(host,port);
  7. //KVStore的实例ID及密码
  8. String authString = jedis.auth(password);//password
  9. if (!authString.equals("OK"))
  10. {
  11. System.err.println("AUTH Failed: " + authString);
  12. return;
  13. }
  14. }
  15. public void pub(String channel,String message){
  16. System.out.println(" >>> 发布(PUBLISH) > Channel:"+channel+" > 发送出的Message:"+message);
  17. jedis.publish(channel, message);
  18. }
  19. public void close(String channel){
  20. System.out.println(" >>> 发布(PUBLISH)结束 > Channel:"+channel+" > Message:quit");
  21. //消息发布者结束发送,即发送一个“quit”消息;
  22. jedis.publish(channel, "quit");
  23. }
  24. }

消息订阅者 (即 subscribe client)

  1. package message.kvstore.aliyun.com;
  2. import redis.clients.jedis.Jedis;
  3. import redis.clients.jedis.JedisPubSub;
  4. public class KVStoreSubClient extends Thread{
  5. private Jedis jedis;
  6. private String channel;
  7. private JedisPubSub listener;
  8. public KVStoreSubClient(String host,int port, String password){
  9. jedis = new Jedis(host,port);
  10. //ApsaraDB for Redis的实例ID及密码
  11. String authString = jedis.auth(password);//kvstore_instance_id:password
  12. if (!authString.equals("OK"))
  13. {
  14. System.err.println("AUTH Failed: " + authString);
  15. return;
  16. }
  17. }
  18. public void setChannelAndListener(JedisPubSub listener,String channel){
  19. this.listener=listener;
  20. this.channel=channel;
  21. }
  22. private void subscribe(){
  23. if(listener==null || channel==null){
  24. System.err.println("Error:SubClient> listener or channel is null");
  25. }
  26. System.out.println(" >>> 订阅(SUBSCRIBE) > Channel:"+channel);
  27. System.out.println();
  28. //接收者在侦听订阅的消息时,将会阻塞进程,直至接收到quit消息(被动方式),或主动取消订阅
  29. jedis.subscribe(listener, channel);
  30. }
  31. public void unsubscribe(String channel){
  32. System.out.println(" >>> 取消订阅(UNSUBSCRIBE) > Channel:"+channel);
  33. System.out.println();
  34. listener.unsubscribe(channel);
  35. }
  36. @Override
  37. public void run() {
  38. try{
  39. System.out.println();
  40. System.out.println("----------订阅消息SUBSCRIBE 开始-------");
  41. subscribe();
  42. System.out.println("----------订阅消息SUBSCRIBE 结束-------");
  43. System.out.println();
  44. }catch(Exception e){
  45. e.printStackTrace();
  46. }
  47. }
  48. }

消息监听者

  1. package message.kvstore.aliyun.com;
  2. import redis.clients.jedis.JedisPubSub;
  3. public class KVStoreMessageListener extends JedisPubSub{
  4. @Override
  5. public void onMessage(String channel, String message) {
  6. System.out.println(" <<< 订阅(SUBSCRIBE)< Channel:" + channel + " >接收到的Message:" + message );
  7. System.out.println();
  8. //当接收到的message为quit时,取消订阅(被动方式)
  9. if(message.equalsIgnoreCase("quit")){
  10. this.unsubscribe(channel);
  11. }
  12. }
  13. @Override
  14. public void onPMessage(String pattern, String channel, String message) {
  15. // TODO Auto-generated method stub
  16. }
  17. @Override
  18. public void onSubscribe(String channel, int subscribedChannels) {
  19. // TODO Auto-generated method stub
  20. }
  21. @Override
  22. public void onUnsubscribe(String channel, int subscribedChannels) {
  23. // TODO Auto-generated method stub
  24. }
  25. @Override
  26. public void onPUnsubscribe(String pattern, int subscribedChannels) {
  27. // TODO Auto-generated method stub
  28. }
  29. @Override
  30. public void onPSubscribe(String pattern, int subscribedChannels) {
  31. // TODO Auto-generated method stub
  32. }
  33. }

示例主程序

  1. package message.kvstore.aliyun.com;
  2. import java.util.UUID;
  3. import redis.clients.jedis.JedisPubSub;
  4. public class KVStorePubSubTest {
  5. //ApsaraDB for Redis的连接信息,从控制台可以获得
  6. static final String host = "xxxxxxxxxx.m.cnhza.kvstore.aliyuncs.com";
  7. static final int port = 6379;
  8. static final String password="xxxxxxxxxx:yyyyyyyy";//kvstore_instance_id:password
  9. public static void main(String[] args) throws Exception{
  10. KVStorePubClient pubClient = new KVStorePubClient(host, port,password);
  11. final String channel = "KVStore频道-A";
  12. //消息发送者开始发消息,此时还无人订阅,所以此消息不会被接收
  13. pubClient.pub(channel, "Aliyun消息1:(此时还无人订阅,所以此消息不会被接收)");
  14. //消息接收者
  15. KVStoreSubClient subClient = new KVStoreSubClient(host, port,password);
  16. JedisPubSub listener = new KVStoreMessageListener();
  17. subClient.setChannelAndListener(listener, channel);
  18. //消息接收者开始订阅
  19. subClient.start();
  20. //消息发送者继续发消息
  21. for (int i = 0; i < 5; i++) {
  22. String message=UUID.randomUUID().toString();
  23. pubClient.pub(channel, message);
  24. Thread.sleep(1000);
  25. }
  26. //消息接收者主动取消订阅
  27. subClient.unsubscribe(channel);
  28. Thread.sleep(1000);
  29. pubClient.pub(channel, "Aliyun消息2:(此时订阅取消,所以此消息不会被接收)");
  30. //消息发布者结束发送,即发送一个“quit”消息;
  31. //此时如果有其他的消息接收者,那么在listener.onMessage()中接收到“quit”时,将执行“unsubscribe”操作。
  32. pubClient.close(channel);
  33. }
  34. }

运行结果

在输入了正确的 ApsaraDB for Redis 实例访问地址和密码之后,运行以上 Java 程序,输出结果如下。

  1. >>> 发布(PUBLISH) > Channel:KVStore频道-A > 发送出的Message:Aliyun消息1:(此时还无人订阅,所以此消息不会被接收)
  2. ----------订阅消息SUBSCRIBE 开始-------
  3. >>> 订阅(SUBSCRIBE) > Channel:KVStore频道-A
  4. >>> 发布(PUBLISH) > Channel:KVStore频道-A > 发送出的Message:0f9c2cee-77c7-4498-89a0-1dc5a2f65889
  5. <<< 订阅(SUBSCRIBE)< Channel:KVStore频道-A >接收到的Message:0f9c2cee-77c7-4498-89a0-1dc5a2f65889
  6. >>> 发布(PUBLISH) > Channel:KVStore频道-A > 发送出的Message:ed5924a9-016b-469b-8203-7db63d06f812
  7. <<< 订阅(SUBSCRIBE)< Channel:KVStore频道-A >接收到的Message:ed5924a9-016b-469b-8203-7db63d06f812
  8. >>> 发布(PUBLISH) > Channel:KVStore频道-A > 发送出的Message:f1f84e0f-8f35-4362-9567-25716b1531cd
  9. <<< 订阅(SUBSCRIBE)< Channel:KVStore频道-A >接收到的Message:f1f84e0f-8f35-4362-9567-25716b1531cd
  10. >>> 发布(PUBLISH) > Channel:KVStore频道-A > 发送出的Message:746bde54-af8f-44d7-8a49-37d1a245d21b
  11. <<< 订阅(SUBSCRIBE)< Channel:KVStore频道-A >接收到的Message:746bde54-af8f-44d7-8a49-37d1a245d21b
  12. >>> 发布(PUBLISH) > Channel:KVStore频道-A > 发送出的Message:8ac3b2b8-9906-4f61-8cad-84fc1f15a3ef
  13. <<< 订阅(SUBSCRIBE)< Channel:KVStore频道-A >接收到的Message:8ac3b2b8-9906-4f61-8cad-84fc1f15a3ef
  14. >>> 取消订阅(UNSUBSCRIBE) > Channel:KVStore频道-A
  15. ----------订阅消息SUBSCRIBE 结束-------
  16. >>> 发布(PUBLISH) > Channel:KVStore频道-A > 发送出的Message:Aliyun消息2:(此时订阅取消,所以此消息不会被接收)
  17. >>> 发布(PUBLISH)结束 > Channel:KVStore频道-A > Message:quit

以上示例中仅演示了一个发布者与一个订阅者的情况,实际上发布者与订阅者都可以为多个,发送消息的频道(channel)也可以是多个,对以上代码稍作修改即可。

最后更新:2016-12-16 17:21:38

  上一篇:go 商品相关性分析__最佳实践_云数据库 Redis 版-阿里云
  下一篇:go 管道传输__最佳实践_云数据库 Redis 版-阿里云