diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/TopicRouteCache.java b/proxy/src/main/java/org/apache/rocketmq/proxy/client/TopicRouteCache.java index 287841dc1b..e013cc4e4e 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/TopicRouteCache.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/client/TopicRouteCache.java @@ -66,7 +66,7 @@ public class TopicRouteCache { protected static MessageQueueWrapper getCacheMessageQueueWrapper(LoadingCache topicCache, String key) throws Exception { MessageQueueWrapper res = topicCache.get(key); - if (res.isEmptyCacheQueue()) { + if (res.isEmptyCachedQueue()) { throw new MQClientException(ResponseCode.TOPIC_NOT_EXIST, "No topic route info in name server for the topic: " + key); } @@ -102,10 +102,10 @@ public class TopicRouteCache { log.info("load {} from namesrv. topic: {}, queue: {}", loaderName(), topic, tmp); return tmp; } - return MessageQueueWrapper.EMPTY_CACHE_QUEUES; + return MessageQueueWrapper.EMPTY_CACHED_QUEUE; } catch (Exception e) { if (RocketMQHelper.isTopicNotExistError(e)) { - return MessageQueueWrapper.EMPTY_CACHE_QUEUES; + return MessageQueueWrapper.EMPTY_CACHED_QUEUE; } throw e; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/route/MessageQueueWrapper.java b/proxy/src/main/java/org/apache/rocketmq/proxy/client/route/MessageQueueWrapper.java index bfffbe31cc..5198d7bbfe 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/route/MessageQueueWrapper.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/client/route/MessageQueueWrapper.java @@ -19,7 +19,7 @@ package org.apache.rocketmq.proxy.client.route; import org.apache.rocketmq.common.protocol.route.TopicRouteData; public class MessageQueueWrapper { - public static final MessageQueueWrapper EMPTY_CACHE_QUEUES = new MessageQueueWrapper("", new TopicRouteData()); + public static final MessageQueueWrapper EMPTY_CACHED_QUEUE = new MessageQueueWrapper("", new TopicRouteData()); private final SelectableMessageQueue read; private final SelectableMessageQueue write; @@ -40,8 +40,8 @@ public class MessageQueueWrapper { return topicRouteWrapper.getTopicName(); } - public boolean isEmptyCacheQueue() { - return this == EMPTY_CACHE_QUEUES; + public boolean isEmptyCachedQueue() { + return this == EMPTY_CACHED_QUEUE; } @Override diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/route/SelectableMessageQueue.java b/proxy/src/main/java/org/apache/rocketmq/proxy/client/route/SelectableMessageQueue.java index 1ed80ff522..390b19a577 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/route/SelectableMessageQueue.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/client/route/SelectableMessageQueue.java @@ -18,50 +18,50 @@ package org.apache.rocketmq.proxy.client.route; import java.util.ArrayList; import java.util.Collections; -import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Random; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.common.constant.PermName; import org.apache.rocketmq.common.message.MessageQueue; import org.apache.rocketmq.common.protocol.route.QueueData; public class SelectableMessageQueue { + private static final int BROKER_ACTING_QUEUE_ID = -1; - // queueId : normal - private final List queues; - // queueId : -1 - private final List brokers; - private final Map brokerNameMap; + // multiple queues for one broker, with queueId : normal + private final List queues = new ArrayList<>(); + // one queue for one broker, with queueId : -1 + private final List brokerActingQueues = new ArrayList<>(); + private final Map brokerNameQueueMap = new ConcurrentHashMap<>(); private final AtomicInteger queueIndex; private final AtomicInteger brokerIndex; public SelectableMessageQueue(TopicRouteWrapper topicRouteWrapper, boolean read) { - this.queues = new ArrayList<>(); - this.brokers = new ArrayList<>(); - this.brokerNameMap = new HashMap<>(); if (read) { this.queues.addAll(buildRead(topicRouteWrapper)); } else { this.queues.addAll(buildWrite(topicRouteWrapper)); } - buildBroker(topicRouteWrapper.getTopicName(), this.queues); + buildBrokerActingQueues(topicRouteWrapper.getTopicName(), this.queues); this.queueIndex = new AtomicInteger(Math.abs(new Random().nextInt())); this.brokerIndex = new AtomicInteger(Math.abs(new Random().nextInt())); } - private static List buildRead( - TopicRouteWrapper topicRoute) { - List queues = new ArrayList<>(); + private static List buildRead(TopicRouteWrapper topicRoute) { + Set queueSet = new HashSet<>(); List qds = topicRoute.getQueueDatas(); if (qds == null) { - return queues; + return new ArrayList<>(); } - Collections.sort(qds); + for (QueueData qd : qds) { if (PermName.isReadable(qd.getPerm())) { String brokerAddr = topicRoute.getMasterAddrPrefer(qd.getBrokerName()); @@ -73,20 +73,16 @@ public class SelectableMessageQueue { AddressableMessageQueue mq = new AddressableMessageQueue( new MessageQueue(topicRoute.getTopicName(), qd.getBrokerName(), i), brokerAddr); - if (!queues.contains(mq)) { - queues.add(mq); - } + queueSet.add(mq); } } } - Collections.sort(queues); - return queues; + return queueSet.stream().sorted().collect(Collectors.toList()); } - private static List buildWrite( - TopicRouteWrapper topicRoute) { - List queues = new ArrayList<>(); + private static List buildWrite(TopicRouteWrapper topicRoute) { + Set queueSet = new HashSet<>(); // order topic route. if (StringUtils.isNotBlank(topicRoute.getOrderTopicConf())) { String[] brokers = topicRoute.getOrderTopicConf().split(";"); @@ -103,61 +99,59 @@ public class SelectableMessageQueue { AddressableMessageQueue mq = new AddressableMessageQueue( new MessageQueue(topicRoute.getTopicName(), brokerName, i), brokerAddr); - if (!queues.contains(mq)) { - queues.add(mq); - } + queueSet.add(mq); } } } else { List qds = topicRoute.getQueueDatas(); if (qds == null) { - return queues; + return new ArrayList<>(); } - Collections.sort(qds); + for (QueueData qd : qds) { if (PermName.isWriteable(qd.getPerm())) { - String brokerName = qd.getBrokerName(); - String brokerAddr = topicRoute.getMasterAddr(brokerName); + String brokerAddr = topicRoute.getMasterAddr(qd.getBrokerName()); if (brokerAddr == null) { continue; } for (int i = 0; i < qd.getWriteQueueNums(); i++) { AddressableMessageQueue mq = new AddressableMessageQueue( - new MessageQueue(topicRoute.getTopicName(), brokerName, i), + new MessageQueue(topicRoute.getTopicName(), qd.getBrokerName(), i), brokerAddr); - if (!queues.contains(mq)) { - queues.add(mq); - } + queueSet.add(mq); } } } } - Collections.sort(queues); - return queues; + return queueSet.stream().sorted().collect(Collectors.toList()); } - private void buildBroker(String topic, List queues) { - for (AddressableMessageQueue messageQueue : queues) { - AddressableMessageQueue mb = new AddressableMessageQueue( - new MessageQueue(topic, messageQueue.getMessageQueue().getBrokerName(), -1), - messageQueue.getBrokerAddr()); - if (!brokers.contains(mb)) { - brokers.add(mb); - brokerNameMap.put(mb.getBrokerName(), mb); + + // 这里应该不是线程安全的 + private void buildBrokerActingQueues(String topic, List normalQueues) { + for (AddressableMessageQueue mq : normalQueues) { + AddressableMessageQueue brokerActingQueue = new AddressableMessageQueue( + new MessageQueue(topic, mq.getMessageQueue().getBrokerName(), BROKER_ACTING_QUEUE_ID), + mq.getBrokerAddr()); + + if (!brokerActingQueues.contains(brokerActingQueue)) { + brokerActingQueues.add(brokerActingQueue); + brokerNameQueueMap.put(brokerActingQueue.getBrokerName(), brokerActingQueue); } } - Collections.sort(brokers); + Collections.sort(brokerActingQueues); } - public final AddressableMessageQueue getBrokerByName(String brokerName) { - return this.brokerNameMap.get(brokerName); + public final AddressableMessageQueue getQueueByBrokerName(String brokerName) { + return this.brokerNameQueueMap.get(brokerName); } public final AddressableMessageQueue selectOne(boolean onlyBroker) { - return selectOneByIndex(onlyBroker ? brokerIndex.getAndIncrement() : queueIndex.getAndIncrement(), onlyBroker); + int nextIndex = onlyBroker ? brokerIndex.getAndIncrement() : queueIndex.getAndIncrement(); + return selectOneByIndex(nextIndex, onlyBroker); } public final AddressableMessageQueue selectOne(String brokerName, int queueId) { @@ -172,22 +166,23 @@ public class SelectableMessageQueue { public final AddressableMessageQueue selectOneByIndex(int index, boolean onlyBroker) { if (onlyBroker) { - if (brokers.isEmpty()) { + if (brokerActingQueues.isEmpty()) { return null; } - return brokers.get(Math.abs(index) % brokers.size()); + return brokerActingQueues.get(Math.abs(index) % brokerActingQueues.size()); } + if (queues.isEmpty()) { return null; } return queues.get(Math.abs(index) % queues.size()); } - public final AddressableMessageQueue selectNextOne( - AddressableMessageQueue last) { + // find next same type(but different) queue with last(normal queue or broker acting queue). + public final AddressableMessageQueue selectNextQueue(AddressableMessageQueue last) { boolean onlyBroker = last.getQueueId() < 0; AddressableMessageQueue newOne = last; - int count = onlyBroker ? brokers.size() : queues.size(); + int count = onlyBroker ? brokerActingQueues.size() : queues.size(); for (int i = 0; i < count; i++) { newOne = selectOne(onlyBroker); @@ -199,21 +194,21 @@ public class SelectableMessageQueue { return newOne; } - public final AddressableMessageQueue selectNextBrokerOne( - AddressableMessageQueue last) { - boolean onlyBroker = last.getQueueId() < 0; - AddressableMessageQueue newOne = last; - int count = onlyBroker ? brokers.size() : queues.size(); - - for (int i = 0; i < count; i++) { - newOne = selectOne(onlyBroker); - if (!newOne.getBrokerName().equals(last.getBrokerName())) { - break; - } - } - - return newOne; - } + // should use selectNextQueue +// public final AddressableMessageQueue selectNextBrokerActingQueue(AddressableMessageQueue last) { +// boolean onlyBroker = last.getQueueId() < 0; +// AddressableMessageQueue newOne = last; +// int count = onlyBroker ? brokerActingQueues.size() : queues.size(); +// +// for (int i = 0; i < count; i++) { +// newOne = selectOne(onlyBroker); +// if (!newOne.getBrokerName().equals(last.getBrokerName())) { +// break; +// } +// } +// +// return newOne; +// } @Override public boolean equals(Object o) { @@ -225,18 +220,18 @@ public class SelectableMessageQueue { } SelectableMessageQueue queue = (SelectableMessageQueue) o; return Objects.equals(queues, queue.queues) && - Objects.equals(brokers, queue.brokers); + Objects.equals(brokerActingQueues, queue.brokerActingQueues); } @Override public int hashCode() { - return Objects.hash(queues, brokers); + return Objects.hash(queues, brokerActingQueues); } @Override public String toString() { return "SelectableMessageQueue{" + "queues=" + queues + - ", brokers=" + brokers + + ", brokers=" + brokerActingQueues + ", queueIndex=" + queueIndex + ", brokerIndex=" + brokerIndex + '}';