摘要：
一、问题答案是不可以的而且后注册的会替换前注册的，MqConsumer2会替换MqConsumer，并且只结束tag-2的消息/***@date2019/05/28*/@Component@Slf4jpublicclassMqConsumerimplementsMessageConsumer{@Override@Transactional(rollbackFor=Throwable.class,p
一、问题答案
是不可以的
而且后注册的会替换前注册的，MqConsumer2会替换MqConsumer，并且只结束tag-2的消息
/** * @date 2019/05/28 */ @Component @Slf4j public class MqConsumer implementsMessageConsumer { @Override @Transactional(rollbackFor = Throwable.class, propagation =Propagation.REQUIRED) public voidonMessage(String msg) { log.info("接收到的库存MQ消息：{}", msg); log.info("接收到的库存MQ消息：{}", msg); log.info("接收到的库存MQ消息：{}", msg); } @Override publicString getTopic() { return "topic-1"; } @Override publicString getTag() { return "tag-1"; } }
@Component @Slf4j public class MqConsumer2 implementsMessageConsumer { @Override @Transactional(rollbackFor = Throwable.class, propagation =Propagation.REQUIRED) public voidonMessage(String msg) { log.info("接收到的库存MQ消息：{}", msg); log.info("接收到的库存MQ消息：{}", msg); log.info("接收到的库存MQ消息：{}", msg); } @Override publicString getTopic() { return "topic-1"; } @Override publicString getTag() { return "tag-2"; } }
二、为什么呢？
我们从源码的角度来分析下
1.订阅消息的方法 public void subscribe(String topic, String subExpression, MessageListener listener) ，其中subExpression即为tag
packagecom.aliyun.openservices.ons.api.impl.rocketmq; .... @Generated("ons-client") public class ConsumerImpl extends ONSConsumerAbstract implementsConsumer { private final ConcurrentHashMap<String, MessageListener> subscribeTable = new ConcurrentHashMap<String, MessageListener>(); public ConsumerImpl(finalProperties properties) { super(properties); boolean postSubscriptionWhenPull = Boolean.parseBoolean(properties.getProperty(PropertyKeyConst.PostSubscriptionWhenPull, "false")); this.defaultMQPushConsumer.setPostSubscriptionWhenPull(postSubscriptionWhenPull); String messageModel =properties.getProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING); this.defaultMQPushConsumer.setMessageModel(MessageModel.valueOf(messageModel)); } @Override public voidstart() { this.defaultMQPushConsumer.registerMessageListener(newMessageListenerImpl()); super.start(); } @Override public voidsubscribe(String topic, String subExpression, MessageListener listener) { if (null ==topic) { throw new ONSClientException("topic is null"); } if (null ==listener) { throw new ONSClientException("listener is null"); } this.subscribeTable.put(topic, listener); super.subscribe(topic, subExpression); } ..... }
从上面的类中我们可以从this.subscribeTable.put(topic, listener);看到subscribeTable这样的一个Map，该Map与tag无关
2.我们再看super.subscribe(topic, subExpression)方法，属于ONSConsumerAbstract类中
protected voidsubscribe(String topic, String subExpression) { try{ this.defaultMQPushConsumer.subscribe(topic, subExpression); } catch(MQClientException e) { throw new ONSClientException("defaultMQPushConsumer subscribe exception", e); } }
DefaultMQPushConsumer中：
@Override public void subscribe(String topic, String subExpression) throwsMQClientException { this.defaultMQPushConsumerImpl.subscribe(withNamespace(topic), subExpression); }
DefaultMQPushConsumerImpl中：
public void subscribe(String topic, String subExpression) throwsMQClientException { try{
//此处用来构建订阅数据，并且指定了tag SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(), topic, subExpression);
//此处将topic和该topic的订阅数据存放到subscriptionInner这个Map中
// protected final ConcurrentMap<String /* topic */, SubscriptionData> subscriptionInner =new ConcurrentHashMap<String, SubscriptionData>(); this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData); if (this.mQClientFactory != null) { this.mQClientFactory.sendHeartbeatToAllBrokerWithLock(); } } catch(Exception e) { throw new MQClientException("subscription exception", e); } }
三、总结
从上面简单的源码可以看到，有用到两个Map，
subscribeTable 和 subscriptionInner ，并且Map的key都为topic。所以我们可以笃定，RocketMQ在同一个项目中，只支持注册一个topic消费者，那么也就只能指定一个tag