正文
rocketMQ 消息的 tag
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
tag 的使用场景：不同的消费组，订阅同一 topic 不同的 tag，拉取不同的消息并消费。在 topic 内部对消息进行隔离。
producer 发送消息，指定 tag
Message msg = new Message("topic-zhang" /* Topic */,
"TagA" /* Tag */,
"key-zhang",
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);
SendResult sendResult = producer.send(msg);
consumer 订阅 topic，指定 tag
consumer.subscribe("topic-zhang", "TagA||TagB||TagC");
broker 存储 consumer 订阅的 tag 信息
// org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData
private Set<String> tagsSet = new HashSet<String>();
private Set<Integer> codeSet = new HashSet<Integer>();
broker 计算 tag 的 hashCode，直接取字符串的 hashcode 值
// org.apache.rocketmq.store.CommitLog#checkMessageAndReturnSize
long tagsCode = 0;
String tags = propertiesMap.get(MessageConst.PROPERTY_TAGS);
if (tags != null && tags.length() > 0) {
tagsCode = MessageExtBrokerInner.tagsString2tagsCode(MessageExt.parseTopicFilterType(sysFlag), tags);
} public static long tagsString2tagsCode(final TopicFilterType filter, final String tags) {
if (null == tags || tags.length() == 0) { return 0; } return tags.hashCode();
}
broker 存储消息 tag 的 hashcode 值：
consumeQueue 中一个 entry 占 20 字节：8字节 commitLog 物理偏移，4字节消息大小，8字节 tag 的 hashCode 值
consumer 拉取消息时，broker 根据 conusmer 提供的 offset 去遍历 consumeQueue，检查 tag 的 hashcode 值，是否满足 consumer 的订阅信息，来对消息进行过滤
// org.apache.rocketmq.store.DefaultMessageStore#getMessage
if (messageFilter != null
&& !messageFilter.isMatchedByConsumeQueue(isTagsCodeLegal ? tagsCode : null, extRet ? cqExtUnit : null)) {
if (getResult.getBufferTotalSize() == 0) {
status = GetMessageStatus.NO_MATCHED_MESSAGE;
} continue;
} // org.apache.rocketmq.broker.filter.ExpressionMessageFilter#isMatchedByConsumeQueue
public boolean isMatchedByConsumeQueue(Long tagsCode, ConsumeQueueExt.CqExtUnit cqExtUnit) {
if (null == subscriptionData) {
return true;
} if (subscriptionData.isClassFilterMode()) {
return true;
} // by tags code.
if (ExpressionType.isTagType(subscriptionData.getExpressionType())) { if (tagsCode == null) {
return true;
} if (subscriptionData.getSubString().equals(SubscriptionData.SUB_ALL)) {
return true;
} return subscriptionData.getCodeSet().contains(tagsCode.intValue());
} else {
// no expression or no bloom
if (consumerFilterData == null || consumerFilterData.getExpression() == null
|| consumerFilterData.getCompiledExpression() == null || consumerFilterData.getBloomFilterData() == null) {
return true;
} // message is before consumer
if (cqExtUnit == null || !consumerFilterData.isMsgInLive(cqExtUnit.getMsgStoreTime())) {
log.debug("Pull matched because not in live: {}, {}", consumerFilterData, cqExtUnit);
return true;
} byte[] filterBitMap = cqExtUnit.getFilterBitMap();
BloomFilter bloomFilter = this.consumerFilterManager.getBloomFilter();
if (filterBitMap == null || !this.bloomDataValid
|| filterBitMap.length * Byte.SIZE != consumerFilterData.getBloomFilterData().getBitNum()) {
return true;
} BitsArray bitsArray = null;
try {
bitsArray = BitsArray.create(filterBitMap);
boolean ret = bloomFilter.isHit(consumerFilterData.getBloomFilterData(), bitsArray);
log.debug("Pull {} by bit map:{}, {}, {}", ret, consumerFilterData, bitsArray, cqExtUnit);
return ret;
} catch (Throwable e) {
log.error("bloom filter error, sub=" + subscriptionData
+ ", filter=" + consumerFilterData + ", bitMap=" + bitsArray, e);
}
} return true;
}
