tencent cloud

PrevNext

文档反馈

请输入关键字

Recent Pages

文档
消息队列 Pulsar 版
    文档
    Download PDF

    消息标签过滤

    最后更新时间：2024-01-03 14:12:49
    下载PDF
      本文主要介绍 TDMQ Pulsar 版中消息标签过滤的功能、应用场景和使用方式。

      功能介绍

      Tag，即消息标签，用于对某个Topic下的消息进行分类。TDMQ Pulsar 版的生产者在发送消息时，指定消息的 Tag，消费者需根据已经指定的 Tag 来进行订阅。
      消费者订阅 Topic 时若未设置 Tag，Topic 中的所有消息都将被投递到消费端进行消费。
      注意：
      在一个订阅中，单个消费者可以使用多个 Tag，多个 Tag 之间的关系是「或」，不同消费者使用的 Tag 需要是相同的。

      应用场景

      通常，一个 Topic 中存放的是相同业务属性的消息，例如交易流水 Topic 包含了下单流水、支付流水、发货流水等，业务若只想消费者其中一种类别的流水，可在客户端进行过滤，但这种过滤方式会带来带宽的资源浪费。
      针对上述场景，TDMQ Pulsar 提供 Broker 端过滤的方式，用户可在生产消息时设置一个或者多个 Tag 标签，消费时指定 Tag 订阅。
      
      img
      
      

      使用说明

      Tag 消息目前是通过 Properties 的方式传入的，可以通过如下方式获取：
      Java
      Go
      <dependency>
          <groupId>org.apache.pulsar</groupId>
          <artifactId>pulsar-client</artifactId>
          <version>2.10.1</version> <!-- 推荐版本 -->
      </dependency>
      推荐使用 0.8.0 及以上版本。
      go get -u github.com/apache/pulsar-client-go@master

      Tag 消息使用限制

      Tag 消息不支持 Batch 功能，Batch 功能默认是开启的。如果要使用 Tag 消息，需要在 Producer 侧禁用掉 batch，具体如下：
      Java
      Go
        // 构建生产者
        Producer<byte[]> producer = pulsarClient.newProducer()
                // 禁用掉batch功能
                .enableBatching(false)
                // topic完整路径，格式为persistent://集群（租户）ID/命名空间/Topic名称
                .topic("persistent://pulsar-xxx/sdk_java/topic2").create();
        producer, err := client.CreateProducer(pulsar.ProducerOptions{
            DisableBatching: true, // 禁用掉batch功能
        })
      tag 消息的过滤只针对已设置 tag 的消息，未设置 tag 的消息，不在过滤范围内。即未设置 tag 的消息会推送给所有的订阅者。
      如果要开启 Tag 消息，需要发送消息的时候，在 ProducerMessage 中设置 Properties 字段；同时在创建 Consumer 的时候需要在 ConsumerOptions 中指定 SubscriptionProperties 字段。
      在 ProducerMessage 中设置 Properties 字段时，其中 key 为 tag 的名字，value 为固定值：TAGS。
      在 ConsumerOptions 中指定 SubscriptionProperties 字段时，其中 key 为要订阅的 tag 的名字，value 为 tag 的版本信息，为保留字段，目前没有实质含义，用来做后续功能的扩展，具体如下：
      指定单个 tag。
      Java
      Go
       // 发送消息
       MessageId msgId = producer.newMessage()
           .property("tag1", "TAGS")
           .value(value.getBytes(StandardCharsets.UTF_8))
              .send();
      
       // 订阅相关参数，可用来设置订阅标签(TAG)
       HashMap<String, String> subProperties = new HashMap<>();
        subProperties.put("tag1","1");
        // 构建消费者
        Consumer<byte[]> consumer = pulsarClient.newConsumer()
            // topic完整路径，格式为persistent://集群（租户）ID/命名空间/Topic名称，从【Topic管理】处复制
           .topic("persistent://pulsar-xxxx/sdk_java/topic2")
           // 需要在控制台Topic详情页创建好一个订阅，此处填写订阅名
           .subscriptionName("topic_sub1")
           // 声明消费模式为共享模式
           .subscriptionType(SubscriptionType.Shared)
           // 订阅相关参数，tag订阅等。。
           .subscriptionProperties(subProperties)
            // 配置从最早开始消费，否则可能会消费不到历史消息
            .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
         // 发送消息
         if msgId, err := producer.Send(ctx, &pulsar.ProducerMessage{
                    Payload: []byte(fmt.Sprintf("hello-%d", i)),
           Properties: map[string]string{
               "tag1": "TAGS",
           },
         	}); err != nil {
                    log.Fatal(err)
           }
      
      // 创建 consumer
           consumer, err := client.Subscribe(pulsar.ConsumerOptions{
                     Topic:            "topic-1",
                     SubscriptionName: "my-sub",
                     SubscriptionProperties: map[string]string{"tag1": "1"},
            })
      指定多个 tag。
      Java
      Go
        // 发送消息
          MessageId msgId = producer.newMessage()
              .property("tag1", "TAGS")
              .property("tag2", "TAGS")
            .value(value.getBytes(StandardCharsets.UTF_8))
              .send();
      
       // 订阅相关参数，可用来设置订阅标签(TAG)
       HashMap<String, String> subProperties = new HashMap<>();
       subProperties.put("tag1","1");
       subProperties.put("tag2","1");
       // 构建消费者
       Consumer<byte[]> consumer = pulsarClient.newConsumer()
           // topic完整路径，格式为persistent://集群（租户）ID/命名空间/Topic名称，从【Topic管理】处复制
           .topic("persistent://pulsar-xxxx/sdk_java/topic2")
           // 需要在控制台Topic详情页创建好一个订阅，此处填写订阅名
           .subscriptionName("topic_sub1")
           // 声明消费模式为共享模式
           .subscriptionType(SubscriptionType.Shared)
           // 订阅相关参数，tag订阅等。。
           .subscriptionProperties(subProperties)
           // 配置从最早开始消费，否则可能会消费不到历史消息
           .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
        // 创建 producer
        if msgId, err := producer.Send(ctx, &pulsar.ProducerMessage{
                   Payload: []byte(fmt.Sprintf("hello-%d", i)),
                     Properties: map[string]string{
                             "tag1": "TAGS",
                             "tag2": "TAGS",
                   },
           }); err != nil {
                   log.Fatal(err)
           }
      
      // 创建 consumer
           consumer, err := client.Subscribe(pulsar.ConsumerOptions{
                     Topic:            "topic-1",
                     SubscriptionName: "my-sub",
                     SubscriptionProperties: map[string]string{
                               "tag1": "1",
                               "tag2": "1",
                     },
           })
      tag 与 properties 混合。
      Java
      Go
         // 发送消息
         MessageId msgId = producer.newMessage()
             .property("tag1", "TAGS")
             .property("tag2", "TAGS")
             .property("xxx", "yyy")
           .value(value.getBytes(StandardCharsets.UTF_8))
             .send();
      
       // 订阅相关参数，可用来设置订阅标签(TAG)
       HashMap<String, String> subProperties = new HashMap<>();
       subProperties.put("tag1","1");
       subProperties.put("tag2","1");
       // 构建消费者
       Consumer<byte[]> consumer = pulsarClient.newConsumer()
           // topic完整路径，格式为persistent://集群（租户）ID/命名空间/Topic名称，从【Topic管理】处复制
           .topic("persistent://pulsar-xxxx/sdk_java/topic2")
           // 需要在控制台Topic详情页创建好一个订阅，此处填写订阅名
           .subscriptionName("topic_sub1")
           // 声明消费模式为共享模式
           .subscriptionType(SubscriptionType.Shared)
           // 订阅相关参数，tag订阅等。。
           .subscriptionProperties(subProperties)
           // 配置从最早开始消费，否则可能会消费不到历史消息
           .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
        // 创建 producer
        if msgId, err := producer.Send(ctx, &pulsar.ProducerMessage{
                   Payload: []byte(fmt.Sprintf("hello-%d", i)),
                   Properties: map[string]string{
                             "tag1": "TAGS",
                             "tag2": "TAGS",
                             "xxx": "yyy",
                   },
           }); err != nil {
                   log.Fatal(err)
           }
      
      // 创建 consumer
           consumer, err := client.Subscribe(pulsar.ConsumerOptions{
                     Topic:            "topic-1",
                     SubscriptionName: "my-sub",
                     SubscriptionProperties: map[string]string{
                               "tag1": "1",
                               "tag2": "1",
                     },
           })
      注意：
      在 consumer 侧设置 SubscriptionProperties 字段时，一旦设定，这个订阅所处理的 tag 信息是不可变更的。如果需要更换订阅的 tag，可以将当前的订阅先 Unsubscribe 掉，然后再重新创建新的订阅来处理。
      
      联系我们

      联系我们，为您的业务提供专属服务。

      联系我们
      技术支持

      如果你想寻求进一步的帮助，通过工单与我们进行联络。我们提供7x24的工单服务。

      提交工单
      7x24 电话支持
      Copyright © 2013-2024 Tencent Cloud. All Rights Reserved.
      隐私条款服务条款缓存条款