全部产品
存储与CDN 数据库 安全 应用服务 数加·人工智能 数加·大数据基础服务 互联网中间件 视频服务 开发者工具 解决方案 物联网 钉钉智能硬件
消息服务

广播拉取消息模型

更新时间:2017-06-07 13:26:11

背景描述

阿里云消息服务MNS 已经提供队列(queue)和主题(topic)两种模型。其中队列提供的是一对多的共享消息消费模型,采用客户端主动拉取(Pull)模式;主题模型提供一对多的广播消息消费模型,并且采用服务端主动推送(Push)模式。上面两种模型基本能满足我们大多数应用场景。

推送模式的好处是即时性能比较好,但是需要暴露客户端地址来接收服务端的消息推送。有些情况下,比如企业内网,我们无法暴露推送地址,希望改用拉取(Pull)的方式。虽然MNS不直接提供这种消费模型,但是我们可以结合主题和队列来实现一对多的拉取消息消费模型。具体方案如下:

解决方案

让主题将消息先推送到队列,然后由消费者从队列拉取消息。这样既可以做到1对多的广播消息,又不需要暴露消费者的地址;如下图所示:

拉取

接口说明

最新的Java SDK(1.1.5)中的CloudPullTopic 默认支持上述解决方案。其中MNSClient 提供下面两个接口来快速创建CloudPullTopic:

  1. public CloudPullTopic createPullTopic(TopicMeta topicMeta, Vector<String> queueNameList, boolean needCreateQueue, QueueMeta queueMetaTemplate)
  2. public CloudPullTopic createPullTopic(TopicMeta topicMeta, Vector<String> queueNameList)

其中,TopicMeta 是创建topic的meta 设置, queueNameList里指定topic消息推送的队列名列表;needCreateQueue表明queueNameList是否需要创建;queueMetaTemplate是创建queue需要的queue meta 参数设置;

Demo代码

  1. CloudAccount account = new CloudAccount(accessKeyId, accessKeySecret, endpoint);
  2. MNSClient client = account.getMNSClient();
  3. // build consumer name list.
  4. Vector<String> consumerNameList = new Vector<String>();
  5. String consumerName1 = "consumer001";
  6. String consumerName2 = "consumer002";
  7. String consumerName3 = "consumer003";
  8. consumerNameList.add(consumerName1);
  9. consumerNameList.add(consumerName2);
  10. consumerNameList.add(consumerName3);
  11. QueueMeta queueMetaTemplate = new QueueMeta();
  12. queueMetaTemplate.setPollingWaitSeconds(30);
  13. try{
  14. //producer code:
  15. // create pull topic which will send message to 3 queues for consumer.
  16. String topicName = "demo-topic-for-pull";
  17. TopicMeta topicMeta = new TopicMeta();
  18. topicMeta.setTopicName(topicName);
  19. CloudPullTopic pullTopic = client.createPullTopic(topicMeta, consumerNameList, true, queueMetaTemplate);
  20. //publish message and consume message.
  21. String messageBody = "broadcast message to all the consumers:hello the world.";
  22. // if we sent raw message,then should use getMessageBodyAsRawString to parse the message body correctly.
  23. TopicMessage tMessage = new RawTopicMessage();
  24. tMessage.setBaseMessageBody(messageBody);
  25. pullTopic.publishMessage(tMessage);
  26. // consumer code:
  27. //3 consumers receive the message.
  28. CloudQueue queueForConsumer1 = client.getQueueRef(consumerName1);
  29. CloudQueue queueForConsumer2 = client.getQueueRef(consumerName2);
  30. CloudQueue queueForConsumer3 = client.getQueueRef(consumerName3);
  31. Message consumer1Msg = queueForConsumer1.popMessage(30);
  32. if(consumer1Msg != null)
  33. {
  34. System.out.println("consumer1 receive message:" + consumer1Msg.getMessageBodyAsRawString());
  35. }else{
  36. System.out.println("the queue is empty");
  37. }
  38. Message consumer2Msg = queueForConsumer2.popMessage(30);
  39. if(consumer2Msg != null)
  40. {
  41. System.out.println("consumer2 receive message:" + consumer2Msg.getMessageBodyAsRawString());
  42. }else{
  43. System.out.println("the queue is empty");
  44. }
  45. Message consumer3Msg = queueForConsumer3.popMessage(30);
  46. if(consumer3Msg != null)
  47. {
  48. System.out.println("consumer3 receive message:" + consumer3Msg.getMessageBodyAsRawString());
  49. }else{
  50. System.out.println("the queue is empty");
  51. }
  52. // delete the fullTopic.
  53. pullTopic.delete();
  54. }catch(ClientException ce)
  55. {
  56. System.out.println("Something wrong with the network connection between client and MNS service."
  57. + "Please check your network and DNS availablity.");
  58. ce.printStackTrace();
  59. }
  60. catch(ServiceException se)
  61. {
  62. /*you can get more MNS service error code in following link.
  63. https://help.aliyun.com/document_detail/mns/api_reference/error_code/error_code.html?spm=5176.docmns/api_reference/error_code/error_response
  64. */
  65. se.printStackTrace();
  66. }
  67. client.close();
本文导读目录