如何实现一对多拉取消息消费模型

问题背景：

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

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

解决方案：

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

fd522053707462f71e44928315782ab2e1cde430

接口说明：

在MNS最新的 (1.1.5)中的CloudPullTopic 默认支持上述解决方案。其中

MNSClient 提供下面两个接口来快速创建CloudPullTopic

public CloudPullTopic createPullTopic(TopicMeta topicMeta, Vector
    
      queueNameList, boolean needCreateQueue, QueueMeta queueMetaTemplate) public CloudPullTopic createPullTopic(TopicMeta topicMeta, Vector
     
       queueNameList)

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

Demo 代码

CloudAccount account = new CloudAccount(accessKeyId, accessKeySecret, endpoint);        MNSClient client = account.getMNSClient();        // build consumer name list.        Vector
     
       consumerNameList = new Vector
      
       ();        String consumerName1 = "consumer001";        String consumerName2 = "consumer002";        String consumerName3 = "consumer003";        consumerNameList.add(consumerName1);        consumerNameList.add(consumerName2);        consumerNameList.add(consumerName3);        QueueMeta queueMetaTemplate = new QueueMeta();        queueMetaTemplate.setPollingWaitSeconds(30);        try{            //producer code:            // create pull topic which will send message to 3 queues for consumer.            String topicName = "demo-topic-for-pull";            TopicMeta topicMeta = new TopicMeta();            topicMeta.setTopicName(topicName);            CloudPullTopic pullTopic = client.createPullTopic(topicMeta, consumerNameList, true, queueMetaTemplate);            //publish message and consume message.            String messageBody = "broadcast message to all the consumers:hello the world.";            // if we sent raw message,then should use getMessageBodyAsRawString to parse the message body correctly.            TopicMessage tMessage = new RawTopicMessage();             tMessage.setBaseMessageBody(messageBody);            pullTopic.publishMessage(tMessage);            // consumer code:            //3 consumers receive the message.            CloudQueue queueForConsumer1 = client.getQueueRef(consumerName1);            CloudQueue queueForConsumer2 = client.getQueueRef(consumerName2);            CloudQueue queueForConsumer3 = client.getQueueRef(consumerName3);            Message consumer1Msg = queueForConsumer1.popMessage(30);            if(consumer1Msg != null)             {                System.out.println("consumer1 receive message:" + consumer1Msg.getMessageBodyAsRawString());            }else{                System.out.println("the queue is empty");            }            Message consumer2Msg = queueForConsumer2.popMessage(30);            if(consumer2Msg != null)             {                System.out.println("consumer2 receive message:" + consumer2Msg.getMessageBodyAsRawString());            }else{                System.out.println("the queue is empty");            }            Message consumer3Msg = queueForConsumer3.popMessage(30);            if(consumer3Msg != null)             {                System.out.println("consumer3 receive message:" + consumer3Msg.getMessageBodyAsRawString());            }else{                System.out.println("the queue is empty");            }            // delete the fullTopic.            pullTopic.delete();        }catch(ClientException ce)        {            System.out.println("Something wrong with the network connection between client and MNS service."                    + "Please check your network and DNS availablity.");            ce.printStackTrace();        }        catch(ServiceException se)        {            /*you can get more MNS service error code in following link.              https://help.aliyun.com/document_detail/mns/api_reference/error_code/error_code.html?spm=5176.docmns/api_reference/error_code/error_response            */            se.printStackTrace();        }        client.close();