[GitHub] incubator-rocketmq pull request #109: [ROCKETMQ-203]Support client to alloca...

[lizhanhui]

Github user lizhanhui commented on a diff in the pull request:

https://github.com/apache/incubator-rocketmq/pull/109#discussion_r119598779
  
--- 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 allocate(String consumerGroup, String 
currentCID, List mqAll,
+List 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 result = new ArrayList();
+if (!cidAll.contains(currentCID)) {
+log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in 
cidAll: {}",
+consumerGroup,
+currentCID,
+cidAll);
+return result;
+}
+
+//group mq by machine room
+Map mr2Mq = new TreeMap();
+for (MessageQueue mq : mqAll) {
+String brokerMachineRoom = 
machineRoomSelector.brokerDeployIn(mq);
+if (StringUtils.isNoneEmpty(brokerMachineRoom)) {
+if (mr2Mq.get(brokerMachineRoom) == null) {
+mr2Mq.put(brokerMachineRoom, new 
ArrayList());
+}
+mr2Mq.get(brokerMachineRoom).add(mq);
+} else {
+throw new IllegalArgumentException("Machine room is null 
for mq " + mq);
+}
+}
+
+//group consumer by machine room
+Map mr2c = new TreeMap();
+for (String cid : cidAll) {
+String consumerMachineRoom = 
machineRoomSelector.consumerDeployIn(cid);
+if (StringUtils.isNoneEmpty(consumerMachineRoom)) {
+if (mr2c.get(consumerMachineRoom) == null) {
+mr2c.put(consumerMachineRoom, new ArrayList());
+}
+mr2c.get(consumerMachineRoom).add(cid);
+} else {
+throw new IllegalArgumentException("Machine 

[GitHub] incubator-rocketmq pull request #109: [ROCKETMQ-203]Support client to alloca...

[Jaskey]

GitHub user Jaskey opened a pull request:

https://github.com/apache/incubator-rocketmq/pull/109

[ROCKETMQ-203]Support client to allocate message queue in machine room 
nearby priority

JIRA: 
https://issues.apache.org/jira/browse/ROCKETMQ-203?jql=project%20%3D%20ROCKETMQ


You can merge this pull request into a Git repository by running:

$ git pull https://github.com/Jaskey/incubator-rocketmq 
ROCKETMQ-203-machineroom-nearby-strategy

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/incubator-rocketmq/pull/109.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #109


commit fd756669d8f07dc02aca540218013c3f8bd6fbc7
Author: Jaskey 
Date:   2017-05-31T09:04:04Z

Add allocate strategy to support allocating message queues to client  in 
machine room nearby priority




---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---