Github user Jaskey commented on a diff in the pull request:
https://github.com/apache/incubator-rocketmq/pull/109#discussion_r119676071
--- Diff:
client/src/main/java/org/apache/rocketmq/client/consumer/rebalance/AllocateMachineRoomNearby.java
---
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.client.consumer.rebalance;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.client.consumer.AllocateMessageQueueStrategy;
+import org.apache.rocketmq.client.log.ClientLogger;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.slf4j.Logger;
+
+/**
+ * An allocate strategy proxy for based on machine room nearside priority.
An actual allocate strategy can be
+ * specified.
+ *
+ * If any consumer is alive in a machine room, the message queue of the
broker which is deployed in the same machine
+ * should only be allocated to those. Otherwise, those message queues can
be shared along all consumers since there are
+ * no alive consumer to monopolize them.
+ */
+public class AllocateMachineRoomNearby implements
AllocateMessageQueueStrategy {
+ private final Logger log = ClientLogger.getLog();
+
+ private final AllocateMessageQueueStrategy
allocateMessageQueueStrategy;//actual allocate strategy
+ private final MachineRoomSelector machineRoomSelector;
+
+ public AllocateMachineRoomNearby(AllocateMessageQueueStrategy
allocateMessageQueueStrategy,
+ MachineRoomSelector machineRoomSelector) {
+ this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
+ this.machineRoomSelector = machineRoomSelector;
+ }
+
+ @Override
+ public List<MessageQueue> allocate(String consumerGroup, String
currentCID, List<MessageQueue> mqAll,
+ List<String> cidAll) {
+ if (currentCID == null || currentCID.length() < 1) {
+ throw new IllegalArgumentException("currentCID is empty");
+ }
+ if (mqAll == null || mqAll.isEmpty()) {
+ throw new IllegalArgumentException("mqAll is null or mqAll
empty");
+ }
+ if (cidAll == null || cidAll.isEmpty()) {
+ throw new IllegalArgumentException("cidAll is null or cidAll
empty");
+ }
+
+ List<MessageQueue> result = new ArrayList<MessageQueue>();
+ if (!cidAll.contains(currentCID)) {
+ log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in
cidAll: {}",
+ consumerGroup,
+ currentCID,
+ cidAll);
+ return result;
+ }
+
+ //group mq by machine room
+ Map<String, List<MessageQueue>> mr2Mq = new TreeMap<String,
List<MessageQueue>>();
+ for (MessageQueue mq : mqAll) {
+ String brokerMachineRoom =
machineRoomSelector.brokerDeployIn(mq);
+ if (StringUtils.isNoneEmpty(brokerMachineRoom)) {
+ if (mr2Mq.get(brokerMachineRoom) == null) {
+ mr2Mq.put(brokerMachineRoom, new
ArrayList<MessageQueue>());
+ }
+ mr2Mq.get(brokerMachineRoom).add(mq);
+ } else {
+ throw new IllegalArgumentException("Machine room is null
for mq " + mq);
+ }
+ }
+
+ //group consumer by machine room
+ Map<String, List<String>> mr2c = new TreeMap<String,
List<String>>();
+ for (String cid : cidAll) {
+ String consumerMachineRoom =
machineRoomSelector.consumerDeployIn(cid);
+ if (StringUtils.isNoneEmpty(consumerMachineRoom)) {
+ if (mr2c.get(consumerMachineRoom) == null) {
+ mr2c.put(consumerMachineRoom, new ArrayList<String>());
+ }
+ mr2c.get(consumerMachineRoom).add(cid);
+ } else {
+ throw new IllegalArgumentException("Machine room is null
for consumer id " + cid);
+ }
+ }
+
+ List<MessageQueue> allocateResults = new ArrayList<MessageQueue>();
+
+ //1.allocate the mq that deploy in the same machine room with the
current consumer
+ String currentMachineRoom =
machineRoomSelector.consumerDeployIn(currentCID);
+ List<MessageQueue> mqInThisMachineRoom =
mr2Mq.remove(currentMachineRoom);
+ List<String> consumerInThisMachineRoom =
mr2c.get(currentMachineRoom);
+ if (mqInThisMachineRoom != null && !mqInThisMachineRoom.isEmpty())
{
+
allocateResults.addAll(allocateMessageQueueStrategy.allocate(consumerGroup,
currentCID, mqInThisMachineRoom, consumerInThisMachineRoom));
+ }
+
+ //2.allocate the rest mq to each machine room if there are no
consumer alive in that machine room
+ for (String machineRoom : mr2Mq.keySet()) {
+ if (!mr2c.containsKey(machineRoom)) { // no alive consumer in
the corresponding machine room, so all consumers share these queues
+
allocateResults.addAll(allocateMessageQueueStrategy.allocate(consumerGroup,
currentCID, mr2Mq.get(machineRoom), cidAll));
+ }
+ }
+
+ return allocateResults;
+ }
+
+ @Override
+ public String getName() {
+ return "MACHINE_ROOM_NEARBY" + "-" +
allocateMessageQueueStrategy.getName();
+ }
+
+ public interface MachineRoomSelector {
--- End diff --
LGTMï¼I will rename this interface if more guys agree
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
