clean up

dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/loadbalance/AbstractLoadBalance.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 1999-2011 Alibaba Group.
+ *  
+ * Licensed 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 com.alibaba.dubbo.rpc.cluster.loadbalance;
+
+import java.util.List;
+
+import com.alibaba.dubbo.common.Constants;
+import com.alibaba.dubbo.common.URL;
+import com.alibaba.dubbo.rpc.Invoker;
+import com.alibaba.dubbo.rpc.Invocation;
+import com.alibaba.dubbo.rpc.cluster.LoadBalance;
+
+/**
+ * AbstractLoadBalance
+ * 
+ * @author william.liangf
+ */
+public abstract class AbstractLoadBalance implements LoadBalance {
+
+    public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
+        if (invokers == null || invokers.size() == 0)
+            return null;
+        if (invokers.size() == 1)
+            return invokers.get(0);
+        return doSelect(invokers, url, invocation);
+    }
+
+    protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation);
+
+    protected int getWeight(Invoker<?> invoker, Invocation invocation) {
+        int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);
+        if (weight > 0) {
+	        long timestamp = invoker.getUrl().getParameter(Constants.TIMESTAMP_KEY, 0L);
+	    	if (timestamp > 0L) {
+	    		int uptime = (int) (System.currentTimeMillis() - timestamp);
+	    		int warmup = invoker.getUrl().getParameter(Constants.WARMUP_KEY, Constants.DEFAULT_WARMUP);
+	    		if (uptime > 0 && uptime < warmup) {
+	    			weight = calculateWarmupWeight(uptime, warmup, weight);
+	    		}
+	    	}
+        }
+    	return weight;
+    }
+
+    static int calculateWarmupWeight(int uptime, int warmup, int weight) {
+    	int ww = (int) ( (float) uptime / ( (float) warmup / (float) weight ) );
+    	return ww < 1 ? 1 : (ww > weight ? weight : ww);
+    }
+
+}

dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalance.java

@@ -0,0 +1,149 @@
+/*
+ * Copyright 1999-2012 Alibaba Group.
+ *  
+ * Licensed 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 com.alibaba.dubbo.rpc.cluster.loadbalance;
+
+import java.io.UnsupportedEncodingException;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.List;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import com.alibaba.dubbo.common.Constants;
+import com.alibaba.dubbo.common.URL;
+import com.alibaba.dubbo.rpc.Invocation;
+import com.alibaba.dubbo.rpc.Invoker;
+
+/**
+ * ConsistentHashLoadBalance
+ * 
+ * @author william.liangf
+ */
+public class ConsistentHashLoadBalance extends AbstractLoadBalance {
+
+    private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = new ConcurrentHashMap<String, ConsistentHashSelector<?>>();
+
+    @SuppressWarnings("unchecked")
+    @Override
+    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
+        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
+        int identityHashCode = System.identityHashCode(invokers);
+        ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
+        if (selector == null || selector.getIdentityHashCode() != identityHashCode) {
+            selectors.put(key, new ConsistentHashSelector<T>(invokers, invocation.getMethodName(), identityHashCode));
+            selector = (ConsistentHashSelector<T>) selectors.get(key);
+        }
+        return selector.select(invocation);
+    }
+
+    private static final class ConsistentHashSelector<T> {
+
+        private final TreeMap<Long, Invoker<T>> virtualInvokers;
+
+        private final int                       replicaNumber;
+
+        private final int                       identityHashCode;
+
+        private final int[]                     argumentIndex;
+
+        public ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
+            this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
+            this.identityHashCode = System.identityHashCode(invokers);
+            URL url = invokers.get(0).getUrl();
+            this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
+            String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0"));
+            argumentIndex = new int[index.length];
+            for (int i = 0; i < index.length; i ++) {
+                argumentIndex[i] = Integer.parseInt(index[i]);
+            }
+            for (Invoker<T> invoker : invokers) {
+                for (int i = 0; i < replicaNumber / 4; i++) {
+                    byte[] digest = md5(invoker.getUrl().toFullString() + i);
+                    for (int h = 0; h < 4; h++) {
+                        long m = hash(digest, h);
+                        virtualInvokers.put(m, invoker);
+                    }
+                }
+            }
+        }
+
+        public int getIdentityHashCode() {
+            return identityHashCode;
+        }
+
+        public Invoker<T> select(Invocation invocation) {
+            String key = toKey(invocation.getArguments());
+            byte[] digest = md5(key);
+            Invoker<T> invoker = sekectForKey(hash(digest, 0));
+            return invoker;
+        }
+
+        private String toKey(Object[] args) {
+            StringBuilder buf = new StringBuilder();
+            for (int i : argumentIndex) {
+                if (i >= 0 && i < args.length) {
+                    buf.append(args[i]);
+                }
+            }
+            return buf.toString();
+        }
+
+        private Invoker<T> sekectForKey(long hash) {
+            Invoker<T> invoker;
+            Long key = hash;
+            if (!virtualInvokers.containsKey(key)) {
+                SortedMap<Long, Invoker<T>> tailMap = virtualInvokers.tailMap(key);
+                if (tailMap.isEmpty()) {
+                    key = virtualInvokers.firstKey();
+                } else {
+                    key = tailMap.firstKey();
+                }
+            }
+            invoker = virtualInvokers.get(key);
+            return invoker;
+        }
+
+        private long hash(byte[] digest, int number) {
+            return (((long) (digest[3 + number * 4] & 0xFF) << 24)
+                    | ((long) (digest[2 + number * 4] & 0xFF) << 16)
+                    | ((long) (digest[1 + number * 4] & 0xFF) << 8) 
+                    | (digest[0 + number * 4] & 0xFF)) 
+                    & 0xFFFFFFFFL;
+        }
+
+        private byte[] md5(String value) {
+            MessageDigest md5;
+            try {
+                md5 = MessageDigest.getInstance("MD5");
+            } catch (NoSuchAlgorithmException e) {
+                throw new IllegalStateException(e.getMessage(), e);
+            }
+            md5.reset();
+            byte[] bytes = null;
+            try {
+                bytes = value.getBytes("UTF-8");
+            } catch (UnsupportedEncodingException e) {
+                throw new IllegalStateException(e.getMessage(), e);
+            }
+            md5.update(bytes);
+            return md5.digest();
+        }
+
+    }
+
+}

dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/loadbalance/LeastActiveLoadBalance.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 1999-2011 Alibaba Group.
+ *  
+ * Licensed 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 com.alibaba.dubbo.rpc.cluster.loadbalance;
+
+import java.util.List;
+import java.util.Random;
+
+import com.alibaba.dubbo.common.Constants;
+import com.alibaba.dubbo.common.URL;
+import com.alibaba.dubbo.rpc.Invocation;
+import com.alibaba.dubbo.rpc.Invoker;
+import com.alibaba.dubbo.rpc.RpcStatus;
+
+/**
+ * LeastActiveLoadBalance
+ * 
+ * @author william.liangf
+ */
+public class LeastActiveLoadBalance extends AbstractLoadBalance {
+
+    public static final String NAME = "leastactive";
+
+    private final Random random = new Random();
+
+    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
+        int length = invokers.size(); // 总个数
+        int leastActive = -1; // 最小的活跃数
+        int leastCount = 0; // 相同最小活跃数的个数
+        int[] leastIndexs = new int[length]; // 相同最小活跃数的下标
+        int totalWeight = 0; // 总权重
+        int firstWeight = 0; // 第一个权重，用于于计算是否相同
+        boolean sameWeight = true; // 是否所有权重相同
+        for (int i = 0; i < length; i++) {
+        	Invoker<T> invoker = invokers.get(i);
+            int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive(); // 活跃数
+            int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); // 权重
+            if (leastActive == -1 || active < leastActive) { // 发现更小的活跃数，重新开始
+                leastActive = active; // 记录最小活跃数
+                leastCount = 1; // 重新统计相同最小活跃数的个数
+                leastIndexs[0] = i; // 重新记录最小活跃数下标
+                totalWeight = weight; // 重新累计总权重
+                firstWeight = weight; // 记录第一个权重
+                sameWeight = true; // 还原权重相同标识
+            } else if (active == leastActive) { // 累计相同最小的活跃数
+                leastIndexs[leastCount ++] = i; // 累计相同最小活跃数下标
+                totalWeight += weight; // 累计总权重
+                // 判断所有权重是否一样
+                if (sameWeight && i > 0 
+                        && weight != firstWeight) {
+                    sameWeight = false;
+                }
+            }
+        }
+        // assert(leastCount > 0)
+        if (leastCount == 1) {
+            // 如果只有一个最小则直接返回
+            return invokers.get(leastIndexs[0]);
+        }
+        if (! sameWeight && totalWeight > 0) {
+            // 如果权重不相同且权重大于0则按总权重数随机
+            int offsetWeight = random.nextInt(totalWeight);
+            // 并确定随机值落在哪个片断上
+            for (int i = 0; i < leastCount; i++) {
+                int leastIndex = leastIndexs[i];
+                offsetWeight -= getWeight(invokers.get(leastIndex), invocation);
+                if (offsetWeight <= 0)
+                    return invokers.get(leastIndex);
+            }
+        }
+        // 如果权重相同或权重为0则均等随机
+        return invokers.get(leastIndexs[random.nextInt(leastCount)]);
+    }
+}

dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/loadbalance/RandomLoadBalance.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 1999-2011 Alibaba Group.
+ *  
+ * Licensed 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 com.alibaba.dubbo.rpc.cluster.loadbalance;
+
+import java.util.List;
+import java.util.Random;
+
+import com.alibaba.dubbo.common.URL;
+import com.alibaba.dubbo.rpc.Invocation;
+import com.alibaba.dubbo.rpc.Invoker;
+
+/**
+ * random load balance.
+ *
+ * @author qianlei
+ * @author william.liangf
+ */
+public class RandomLoadBalance extends AbstractLoadBalance {
+
+    public static final String NAME = "random";
+
+    private final Random random = new Random();
+
+    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
+        int length = invokers.size(); // 总个数
+        int totalWeight = 0; // 总权重
+        boolean sameWeight = true; // 权重是否都一样
+        for (int i = 0; i < length; i++) {
+            int weight = getWeight(invokers.get(i), invocation);
+            totalWeight += weight; // 累计总权重
+            if (sameWeight && i > 0
+                    && weight != getWeight(invokers.get(i - 1), invocation)) {
+                sameWeight = false; // 计算所有权重是否一样
+            }
+        }
+        if (totalWeight > 0 && ! sameWeight) {
+            // 如果权重不相同且权重大于0则按总权重数随机
+            int offset = random.nextInt(totalWeight);
+            // 并确定随机值落在哪个片断上
+            for (int i = 0; i < length; i++) {
+                offset -= getWeight(invokers.get(i), invocation);
+                if (offset < 0) {
+                    return invokers.get(i);
+                }
+            }
+        }
+        // 如果权重相同或权重为0则均等随机
+        return invokers.get(random.nextInt(length));
+    }
+
+}