[WIP] load balance #3054 (#3057)

* Load balancing abstract

* code
smell

* lower weight select

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/assign/AbstractSelector.java

+/*
+ * 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.dolphinscheduler.server.master.dispatch.host.assign;
+
+import org.apache.dolphinscheduler.common.utils.CollectionUtils;
+
+import java.util.Collection;
+
+/**
+ *  AbstractSelector
+ */
+public  abstract class AbstractSelector<T> implements Selector<T>{
+    @Override
+    public T select(Collection<T> source) {
+
+        if (CollectionUtils.isEmpty(source)) {
+            throw new IllegalArgumentException("Empty source.");
+        }
+
+        /**
+         * if only one , return directly
+         */
+        if (source.size() == 1) {
+            return (T)source.toArray()[0];
+        }
+        return doSelect(source);
+    }
+
+    protected abstract T  doSelect(Collection<T> source);
+
+}

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/assign/LowerWeightRoundRobin.java

@@ -22,7 +22,7 @@ import java.util.Collection;
 /**
  *  lower weight round robin
  */
-public class LowerWeightRoundRobin implements Selector<HostWeight>{
+public class LowerWeightRoundRobin extends AbstractSelector<HostWeight>{
 
     /**
      * select
@@ -30,7 +30,7 @@ public class LowerWeightRoundRobin implements Selector<HostWeight>{
      * @return HostWeight
      */
     @Override
-    public HostWeight select(Collection<HostWeight> sources){
+    public HostWeight doSelect(Collection<HostWeight> sources){
         int totalWeight = 0;
         int lowWeight = 0;
         HostWeight lowerNode = null;

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/assign/RandomSelector.java

@@ -24,23 +24,12 @@ import java.util.Random;
  * random selector
  * @param <T> T
  */
-public class RandomSelector<T> implements Selector<T> {
+public class RandomSelector<T> extends AbstractSelector<T> {
 
     private final Random random = new Random();
 
     @Override
-    public T select(final Collection<T> source) {
-
-        if (source == null || source.size() == 0) {
-            throw new IllegalArgumentException("Empty source.");
-        }
-
-        /**
-         * if only one , return directly
-         */
-        if (source.size() == 1) {
-            return (T) source.toArray()[0];
-        }
+    public T doSelect(final Collection<T> source) {
 
         int size = source.size();
         /**

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/assign/RoundRobinSelector.java

@@ -26,22 +26,12 @@ import java.util.concurrent.atomic.AtomicInteger;
  * @param <T> T
  */
 @Service
-public class RoundRobinSelector<T> implements Selector<T> {
+public class RoundRobinSelector<T> extends AbstractSelector<T> {
 
     private final AtomicInteger index = new AtomicInteger(0);
 
     @Override
-    public T select(Collection<T> source) {
-        if (source == null || source.size() == 0) {
-            throw new IllegalArgumentException("Empty source.");
-        }
-
-        /**
-         * if only one , return directly
-         */
-        if (source.size() == 1) {
-            return (T)source.toArray()[0];
-        }
+    public T doSelect(Collection<T> source) {
 
         int size = source.size();
         /**