注冊一個種監聽消費代碼如下，定義了一個@Configuration完成監聽器注冊
但是我有個需求場景是同時需要消費有序和無序的topic，我嘗試創建兩個Configuration但是只能初始化其中一個，如果我只創建一個Configuration代碼，在代碼函式DefaultMQPushConsumer種 new 兩個 DefaultMQPushConsumer 一個注冊并發listener并start 另一個注冊有序listener并start
啟動還真不報錯，但是實際測驗只能消費兩者后一個注冊的監聽器，請問我怎么寫代碼能做到同時監聽MessageListenerOrderly和MessageListenerConcurrently
@Bean
public DefaultMQPushConsumer defaultConsumer() throws MQClientException {
log.info("創建RocketMQ并發監聽器");
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
consumer.setNamesrvAddr(namesrvAddr);
consumer.setConsumeThreadMin(consumeThreadMin);
consumer.setConsumeThreadMax(consumeThreadMax);
consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
// 設定監聽
consumer.registerMessageListener(consumeMsgListenerProcessor);
/**
* 設定consumer第一次啟動是從佇列頭部開始還是佇列尾部開始
* 如果不是第一次啟動，那么按照上次消費的位置繼續消費
*/
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
/**
* 設定消費模型，集群還是廣播，默認為集群
*/
// consumer.setMessageModel(MessageModel.CLUSTERING);
try {
// 設定該消費者訂閱的主題和tag，如果訂閱該主題下的所有tag，則使用*,
String[] topicArr = topics.split(";");
for (String tag : topicArr) {
String[] tagArr = tag.split("~");
consumer.subscribe(tagArr[0], tagArr[1]);
}
consumer.start();
} catch (MQClientException e) {
log.error("consumer 創建失敗!");
}
return consumer;
}
同時注冊兩種監聽器
@Bean
public DefaultMQPushConsumer defaultConsumer() throws MQClientException {
log.info("創建RocketMQ亂序監聽器");
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
consumer.setNamesrvAddr(namesrvAddr);
consumer.setConsumeThreadMin(consumeThreadMin);
consumer.setConsumeThreadMax(consumeThreadMax);
consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
// 設定監聽
consumer.registerMessageListener(consumeMsgListenerProcessor);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
try {
// 設定該消費者訂閱的主題和tag，如果訂閱該主題下的所有tag，則使用*,
String[] topicArr = topics.split(";");
for (String tag : topicArr) {
String[] tagArr = tag.split("~");
consumer.subscribe(tagArr[0], tagArr[1]);
}
consumer.start();
} catch (MQClientException e) {
log.error("consumer 創建失敗!");
}
log.info("consumer 亂序創建成功 groupName={}, topics={}, namesrvAddr={}", groupName, topics, namesrvAddr);
log.info("創建RocketMQ有序監聽器");
DefaultMQPushConsumer consumerOrderly = new DefaultMQPushConsumer(orderlyGroupName);
consumerOrderly.setNamesrvAddr(namesrvAddr);
consumerOrderly.setConsumeThreadMin(consumeThreadMin);
consumerOrderly.setConsumeThreadMax(consumeThreadMax);
consumerOrderly.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
// 設定監聽
consumerOrderly.registerMessageListener(consumeMsgOrderLyListenerProcessor);
consumerOrderly.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
try {
// 設定該消費者訂閱的主題和tag，如果訂閱該主題下的所有tag，則使用*,
String[] topicArrOrderly = orderlyTopics.split(";");
for (String tag : topicArrOrderly) {
String[] tagArr = tag.split("~");
consumerOrderly.subscribe(tagArr[0], tagArr[1]);
}
consumerOrderly.start();
log.info("consumer 有序創建成功 groupName={}, topics={}, namesrvAddr={}", orderlyGroupName, orderlyTopics, namesrvAddr);
} catch (MQClientException e) {
log.error("consumer 創建失敗!,{}", e);
}
return consumer;
}
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/273443.html
標籤：Web 開發
上一篇：尋求幫助