本文主要介绍 TDMQ Pulsar 版中消息标签过滤的功能、应用场景和使用方式。
功能介绍
Tag,即消息标签,用于对某个Topic下的消息进行分类。TDMQ Pulsar 版的生产者在发送消息时,指定消息的 Tag,消费者需根据已经指定的 Tag 来进行订阅。
消费者订阅 Topic 时若未设置 Tag,Topic 中的所有消息都将被投递到消费端进行消费。
注意
在一个订阅中,单个消费者可以使用多个 Tag,多个 Tag 之间的关系是「或」,不同消费者使用的 Tag 需要是相同的。
应用场景
通常,一个 Topic 中存放的是相同业务属性的消息,例如交易流水 Topic 包含了下单流水、支付流水、发货流水等,业务若只想消费者其中一种类别的流水,可在客户端进行过滤,但这种过滤方式会带来带宽的资源浪费。
针对上述场景,TDMQ Pulsar 提供 Broker 端过滤的方式,用户可在生产消息时设置一个或者多个 Tag 标签,消费时指定 Tag 订阅。
使用说明
Tag 消息目前是通过 Properties 的方式传入的,可以通过如下方式获取:
<dependency><groupId>org.apache.pulsar</groupId><artifactId>pulsar-client</artifactId><version>2.10.1</version> <!-- 推荐版本 --></dependency>
推荐使用 0.9.0 及以上版本。
go get -u github.com/apache/pulsar-client-go@master
Tag 消息使用限制
Tag 消息不支持 Batch 功能,Batch 功能默认是开启的。如果要使用 Tag 消息,需要在 Producer 侧禁用掉 batch,具体如下:
// 构建生产者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
// 发送消息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)}// 创建 consumerconsumer, err := client.Subscribe(pulsar.ConsumerOptions{Topic: "topic-1",SubscriptionName: "my-sub",SubscriptionProperties: map[string]string{"tag1": "1"},})
指定多个 tag
// 发送消息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();
// 创建 producerif 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)}// 创建 consumerconsumer, err := client.Subscribe(pulsar.ConsumerOptions{Topic: "topic-1",SubscriptionName: "my-sub",SubscriptionProperties: map[string]string{"tag1": "1","tag2": "1",},})
tag 与 properties 混合
// 发送消息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();
// 创建 producerif 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)}// 创建 consumerconsumer, err := client.Subscribe(pulsar.ConsumerOptions{Topic: "topic-1",SubscriptionName: "my-sub",SubscriptionProperties: map[string]string{"tag1": "1","tag2": "1",},})
注意
在 consumer 侧设置 SubscriptionProperties 字段时,一旦设定,这个订阅所处理的 tag 信息是不可变更的。如果需要更换订阅的 tag,可以将当前的订阅先
Unsubscribe
掉,然后再重新创建新的订阅来处理。