本文介绍如何使用消息服务MNS实现一对多拉取消息消费模型,以满足一对多订阅、主动拉取的场景。

背景信息

消息服务MNS提供队列(Queue)和主题(Topic)两种模型,基本能满足大多数应用场景。
  • 队列提供的是一对一的共享消息消费模型,采用客户端主动拉取(Pull)模式。
  • 主题模型提供一对多的广播消息消费模型,采用服务端主动推送(Push)模式。

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

解决方案

通过创建订阅,让主题将消息先推送到队列,然后由消费者从队列拉取消息。这样既可以做到一对多的广播消息,又可避免暴露消费者的地址。消息流

接口说明

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

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

public CloudPullTopic createPullTopic(TopicMeta topicMeta, Vector<String> queueNameList)
参数说明如下:
  • TopicMeta:创建主题的参数设置。
  • QueueMeta:创建队列的参数设置。
  • queueNameList:指定主题消息推送的队列名列表。
  • needCreateQueuequeueNameList是否需要创建。
  • queueMetaTemplate:创建队列需要的QueueMeta参数示例。

示例代码

广播拉取消息的示例代码如下:

CloudAccount account = new CloudAccount(accessKeyId, accessKeySecret, endpoint);
        MNSClient client = account.getMNSClient();

        // 创建消费者列表。
        Vector<String> consumerNameList = new Vector<String>();
        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{
           
            // 创建主题。
            String topicName = "demo-topic-for-pull";
            TopicMeta topicMeta = new TopicMeta();
            topicMeta.setTopicName(topicName);
            CloudPullTopic pullTopic = client.createPullTopic(topicMeta, consumerNameList, true, queueMetaTemplate);

            // 发布消息。
            String messageBody = "broadcast message to all the consumers:hello the world.";
            // 如果发送原始消息,使用getMessageBodyAsRawString解析消息体。
            TopicMessage tMessage = new RawTopicMessage(); 
            tMessage.setBaseMessageBody(messageBody);
            pullTopic.publishMessage(tMessage);

            
            // 接收消息。
            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");
            }

            // 删除主题。
            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)
        {
            // 更多错误码信息,请参见错误码。
            se.printStackTrace();
        }

        client.close();