CalvinKirs
3 years ago
32 changed files with 347 additions and 651 deletions
@ -1,17 +0,0 @@
|
||||
package org.apache.dolphinscheduler.remote.rpc.client; |
||||
|
||||
import org.apache.dolphinscheduler.remote.rpc.Invoker; |
||||
import org.apache.dolphinscheduler.remote.rpc.common.RpcRequest; |
||||
import org.apache.dolphinscheduler.remote.rpc.common.RpcResponse; |
||||
|
||||
/** |
||||
* ConsumerInvoker |
||||
*/ |
||||
public class ConsumerInvoker implements Invoker { |
||||
@Override |
||||
public RpcResponse invoke(RpcRequest req) throws Throwable { |
||||
|
||||
System.out.println(req.getRequestId()+"kris"); |
||||
return null; |
||||
} |
||||
} |
||||
@ -1,59 +0,0 @@
|
||||
package org.apache.dolphinscheduler.remote.rpc.directory; |
||||
|
||||
import org.apache.dolphinscheduler.remote.rpc.filter.SelectorFilter; |
||||
|
||||
import java.util.ArrayList; |
||||
import java.util.Collections; |
||||
import java.util.List; |
||||
import java.util.concurrent.ConcurrentHashMap; |
||||
|
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
/** |
||||
* Directory |
||||
*/ |
||||
public class Directory { |
||||
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(Directory.class); |
||||
|
||||
|
||||
private SelectorFilter selectorFilter = SelectorFilter.getInstance(); |
||||
|
||||
public static Directory getInstance() { |
||||
return Directory.DirectoryInner.INSTANCE; |
||||
} |
||||
|
||||
private static class DirectoryInner { |
||||
|
||||
private static final Directory INSTANCE = new Directory(); |
||||
} |
||||
|
||||
private Directory() { |
||||
} |
||||
|
||||
|
||||
private ConcurrentHashMap<String, List<String>> directoryMap = new ConcurrentHashMap<>(); |
||||
|
||||
public List<String> getDirectory(String serviceName) { |
||||
return directoryMap.get(serviceName); |
||||
} |
||||
|
||||
public boolean addServer(String serviceName, String servicePath) { |
||||
synchronized (this) { |
||||
if (directoryMap.containsKey(serviceName)) { |
||||
directoryMap.get(serviceName).add(servicePath); |
||||
return true; |
||||
} |
||||
} |
||||
directoryMap.putIfAbsent(serviceName, new ArrayList<>(Collections.singletonList(servicePath))); |
||||
return true; |
||||
} |
||||
|
||||
public boolean removeServer(String serviceName, String servicePath) { |
||||
|
||||
return true; |
||||
} |
||||
|
||||
} |
||||
@ -1,15 +0,0 @@
|
||||
package org.apache.dolphinscheduler.remote.rpc.filter; |
||||
|
||||
import org.apache.dolphinscheduler.remote.rpc.Invoker; |
||||
import org.apache.dolphinscheduler.remote.rpc.common.RpcRequest; |
||||
import org.apache.dolphinscheduler.remote.rpc.common.RpcResponse; |
||||
|
||||
|
||||
import com.amazonaws.Response; |
||||
|
||||
|
||||
public interface Filter { |
||||
|
||||
|
||||
RpcResponse filter(Invoker invoker, RpcRequest req) throws Throwable; |
||||
} |
||||
@ -1,38 +0,0 @@
|
||||
package org.apache.dolphinscheduler.remote.rpc.filter; |
||||
|
||||
import org.apache.dolphinscheduler.remote.rpc.Invoker; |
||||
|
||||
import java.util.List; |
||||
|
||||
/** |
||||
* FilterChain |
||||
*/ |
||||
public class FilterChain { |
||||
|
||||
|
||||
private List<Filter> filters; |
||||
|
||||
private Invoker invoker; |
||||
|
||||
|
||||
public FilterChain(List<Filter> filters, Invoker invoker) { |
||||
this.filters = filters; |
||||
this.invoker = invoker; |
||||
} |
||||
|
||||
public FilterChain(Invoker invoker) { |
||||
this(LoaderFilters.create().getFilters(), invoker); |
||||
} |
||||
|
||||
public Invoker buildFilterChain() { |
||||
// 最后一个
|
||||
Invoker last = invoker; |
||||
|
||||
for (int i = filters.size() - 1; i >= 0; i--) { |
||||
last = new FilterWrapper(filters.get(i), last); |
||||
} |
||||
// 第一个
|
||||
return last; |
||||
|
||||
} |
||||
} |
||||
@ -1,33 +0,0 @@
|
||||
package org.apache.dolphinscheduler.remote.rpc.filter; |
||||
|
||||
import org.apache.dolphinscheduler.remote.rpc.Invoker; |
||||
import org.apache.dolphinscheduler.remote.rpc.common.RpcRequest; |
||||
import org.apache.dolphinscheduler.remote.rpc.common.RpcResponse; |
||||
import org.apache.dolphinscheduler.remote.utils.Host; |
||||
|
||||
/** |
||||
* @author jiangli |
||||
* @date 2021-01-11 11:48 |
||||
*/ |
||||
public class FilterWrapper implements Invoker { |
||||
|
||||
|
||||
private Filter next; |
||||
|
||||
private Invoker invoker; |
||||
|
||||
|
||||
public FilterWrapper(Filter next, Invoker invoker) { |
||||
this.next = next; |
||||
this.invoker = invoker; |
||||
} |
||||
|
||||
@Override |
||||
public RpcResponse invoke(RpcRequest args) throws Throwable { |
||||
if (next != null) { |
||||
return next.filter(invoker, args); |
||||
} else { |
||||
return invoker.invoke(args); |
||||
} |
||||
} |
||||
} |
||||
@ -1,26 +0,0 @@
|
||||
package org.apache.dolphinscheduler.remote.rpc.filter; |
||||
|
||||
import java.util.ArrayList; |
||||
import java.util.List; |
||||
|
||||
/** |
||||
* LoaderFilters |
||||
*/ |
||||
public class LoaderFilters { |
||||
|
||||
|
||||
private List<Filter> filterList = new ArrayList<>(); |
||||
|
||||
private LoaderFilters() { |
||||
} |
||||
|
||||
public static LoaderFilters create() { |
||||
|
||||
return new LoaderFilters(); |
||||
} |
||||
|
||||
public List<Filter> getFilters() { |
||||
filterList.add(SelectorFilter.getInstance()); |
||||
return filterList; |
||||
} |
||||
} |
||||
@ -1,60 +0,0 @@
|
||||
package org.apache.dolphinscheduler.remote.rpc.filter; |
||||
|
||||
import org.apache.dolphinscheduler.remote.rpc.Invoker; |
||||
import org.apache.dolphinscheduler.remote.rpc.common.RpcRequest; |
||||
import org.apache.dolphinscheduler.remote.rpc.common.RpcResponse; |
||||
import org.apache.dolphinscheduler.remote.rpc.directory.Directory; |
||||
import org.apache.dolphinscheduler.remote.rpc.selector.RandomSelector; |
||||
import org.apache.dolphinscheduler.remote.utils.Host; |
||||
|
||||
|
||||
import java.util.ArrayList; |
||||
import java.util.List; |
||||
|
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
|
||||
/** |
||||
* SelectorFilter |
||||
*/ |
||||
public class SelectorFilter implements Filter { |
||||
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(SelectorFilter.class); |
||||
|
||||
|
||||
private SelectorFilter selectorFilter = SelectorFilter.getInstance(); |
||||
|
||||
public static SelectorFilter getInstance() { |
||||
return SelectorFilterInner.INSTANCE; |
||||
} |
||||
|
||||
|
||||
private static class SelectorFilterInner { |
||||
|
||||
private static final SelectorFilter INSTANCE = new SelectorFilter(); |
||||
} |
||||
|
||||
private SelectorFilter() { |
||||
} |
||||
|
||||
@Override |
||||
public RpcResponse filter(Invoker invoker, RpcRequest req) throws Throwable { |
||||
Directory.getInstance().addServer("default","127.0.0.1:8080"); |
||||
Directory.getInstance().addServer("default","127.0.0.2:8080"); |
||||
Directory.getInstance().addServer("default","127.0.0.3:8080"); |
||||
List<String> hosts = Directory.getInstance().getDirectory("default"); |
||||
List<Host> candidateHosts = new ArrayList<>(hosts.size()); |
||||
hosts.forEach(node -> { |
||||
Host nodeHost = Host.of(node); |
||||
nodeHost.setWorkGroup("default"); |
||||
candidateHosts.add(nodeHost); |
||||
}); |
||||
RandomSelector randomSelector = new RandomSelector(); |
||||
System.out.println(randomSelector.doSelect(candidateHosts)); |
||||
RpcResponse rsp = new RpcResponse(); |
||||
rsp.setMsg("ms"); |
||||
return rsp; |
||||
} |
||||
} |
||||
@ -1,87 +0,0 @@
|
||||
/* |
||||
* 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.remote.rpc.filter.selector; |
||||
|
||||
import org.apache.dolphinscheduler.remote.utils.Constants; |
||||
import org.apache.dolphinscheduler.remote.utils.Host; |
||||
|
||||
/** |
||||
* host weight |
||||
*/ |
||||
public class HostWeight { |
||||
|
||||
private final int CPU_FACTOR = 10; |
||||
|
||||
private final int MEMORY_FACTOR = 20; |
||||
|
||||
private final int LOAD_AVERAGE_FACTOR = 70; |
||||
|
||||
private final Host host; |
||||
|
||||
private final double weight; |
||||
|
||||
private double currentWeight; |
||||
|
||||
public HostWeight(Host host, double cpu, double memory, double loadAverage) { |
||||
this.weight = getWeight(cpu, memory, loadAverage, host); |
||||
this.host = host; |
||||
this.currentWeight = weight; |
||||
} |
||||
|
||||
public double getCurrentWeight() { |
||||
return currentWeight; |
||||
} |
||||
|
||||
public double getWeight() { |
||||
return weight; |
||||
} |
||||
|
||||
public void setCurrentWeight(double currentWeight) { |
||||
this.currentWeight = currentWeight; |
||||
} |
||||
|
||||
public Host getHost() { |
||||
return host; |
||||
} |
||||
|
||||
@Override |
||||
public String toString() { |
||||
return "HostWeight{" |
||||
+ "host=" + host |
||||
+ ", weight=" + weight |
||||
+ ", currentWeight=" + currentWeight |
||||
+ '}'; |
||||
} |
||||
|
||||
private double getWeight(double cpu, double memory, double loadAverage, Host host) { |
||||
double calculateWeight = cpu * CPU_FACTOR + memory * MEMORY_FACTOR + loadAverage * LOAD_AVERAGE_FACTOR; |
||||
return getWarmUpWeight(host, calculateWeight); |
||||
} |
||||
|
||||
/** |
||||
* If the warm-up is not over, add the weight |
||||
*/ |
||||
private double getWarmUpWeight(Host host, double weight) { |
||||
long startTime = host.getStartTime(); |
||||
long uptime = System.currentTimeMillis() - startTime; |
||||
if (uptime > 0 && uptime < Constants.WARM_UP_TIME) { |
||||
return weight * Constants.WARM_UP_TIME / uptime; |
||||
} |
||||
return weight; |
||||
} |
||||
} |
||||
@ -1,56 +0,0 @@
|
||||
/* |
||||
* 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.remote.rpc.filter.selector; |
||||
|
||||
|
||||
import org.apache.dolphinscheduler.remote.rpc.selector.AbstractSelector; |
||||
|
||||
import java.util.Collection; |
||||
|
||||
/** |
||||
* lower weight round robin |
||||
*/ |
||||
public class LowerWeightRoundRobin extends AbstractSelector<HostWeight> { |
||||
|
||||
/** |
||||
* select |
||||
* |
||||
* @param sources sources |
||||
* @return HostWeight |
||||
*/ |
||||
@Override |
||||
public HostWeight doSelect(Collection<HostWeight> sources) { |
||||
double totalWeight = 0; |
||||
double lowWeight = 0; |
||||
HostWeight lowerNode = null; |
||||
for (HostWeight hostWeight : sources) { |
||||
totalWeight += hostWeight.getWeight(); |
||||
hostWeight.setCurrentWeight(hostWeight.getCurrentWeight() + hostWeight.getWeight()); |
||||
if (lowerNode == null || lowWeight > hostWeight.getCurrentWeight()) { |
||||
lowerNode = hostWeight; |
||||
lowWeight = hostWeight.getCurrentWeight(); |
||||
} |
||||
} |
||||
lowerNode.setCurrentWeight(lowerNode.getCurrentWeight() + totalWeight); |
||||
return lowerNode; |
||||
|
||||
} |
||||
} |
||||
|
||||
|
||||
|
||||
@ -1,141 +0,0 @@
|
||||
/* |
||||
* 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.remote.rpc.filter.selector; |
||||
|
||||
import org.apache.dolphinscheduler.remote.rpc.selector.AbstractSelector; |
||||
import org.apache.dolphinscheduler.remote.utils.Host; |
||||
|
||||
import java.util.ArrayList; |
||||
import java.util.Collection; |
||||
import java.util.List; |
||||
import java.util.concurrent.ConcurrentHashMap; |
||||
import java.util.concurrent.ConcurrentMap; |
||||
import java.util.concurrent.atomic.AtomicBoolean; |
||||
import java.util.concurrent.atomic.AtomicLong; |
||||
|
||||
|
||||
/** |
||||
* Smooth Weight Round Robin |
||||
*/ |
||||
public class RoundRobinSelector extends AbstractSelector<Host> { |
||||
|
||||
private ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> workGroupWeightMap = new ConcurrentHashMap<>(); |
||||
|
||||
private static final int RECYCLE_PERIOD = 100000; |
||||
|
||||
private AtomicBoolean updateLock = new AtomicBoolean(); |
||||
|
||||
protected static class WeightedRoundRobin { |
||||
private int weight; |
||||
private AtomicLong current = new AtomicLong(0); |
||||
private long lastUpdate; |
||||
|
||||
int getWeight() { |
||||
return weight; |
||||
} |
||||
|
||||
void setWeight(int weight) { |
||||
this.weight = weight; |
||||
current.set(0); |
||||
} |
||||
|
||||
long increaseCurrent() { |
||||
return current.addAndGet(weight); |
||||
} |
||||
|
||||
void sel(int total) { |
||||
current.addAndGet(-1L * total); |
||||
} |
||||
|
||||
long getLastUpdate() { |
||||
return lastUpdate; |
||||
} |
||||
|
||||
void setLastUpdate(long lastUpdate) { |
||||
this.lastUpdate = lastUpdate; |
||||
} |
||||
|
||||
} |
||||
|
||||
|
||||
@Override |
||||
public Host doSelect(Collection<Host> source) { |
||||
|
||||
List<Host> hosts = new ArrayList<>(source); |
||||
String key = hosts.get(0).getWorkGroup(); |
||||
ConcurrentMap<String, WeightedRoundRobin> map = workGroupWeightMap.get(key); |
||||
if (map == null) { |
||||
workGroupWeightMap.putIfAbsent(key, new ConcurrentHashMap<>()); |
||||
map = workGroupWeightMap.get(key); |
||||
} |
||||
|
||||
int totalWeight = 0; |
||||
long maxCurrent = Long.MIN_VALUE; |
||||
long now = System.currentTimeMillis(); |
||||
Host selectedHost = null; |
||||
WeightedRoundRobin selectWeightRoundRobin = null; |
||||
|
||||
for (Host host : hosts) { |
||||
String workGroupHost = host.getWorkGroup() + host.getAddress(); |
||||
WeightedRoundRobin weightedRoundRobin = map.get(workGroupHost); |
||||
int weight = host.getWeight(); |
||||
if (weight < 0) { |
||||
weight = 0; |
||||
} |
||||
|
||||
if (weightedRoundRobin == null) { |
||||
weightedRoundRobin = new WeightedRoundRobin(); |
||||
// set weight
|
||||
weightedRoundRobin.setWeight(weight); |
||||
map.putIfAbsent(workGroupHost, weightedRoundRobin); |
||||
weightedRoundRobin = map.get(workGroupHost); |
||||
} |
||||
if (weight != weightedRoundRobin.getWeight()) { |
||||
weightedRoundRobin.setWeight(weight); |
||||
} |
||||
|
||||
long cur = weightedRoundRobin.increaseCurrent(); |
||||
weightedRoundRobin.setLastUpdate(now); |
||||
if (cur > maxCurrent) { |
||||
maxCurrent = cur; |
||||
selectedHost = host; |
||||
selectWeightRoundRobin = weightedRoundRobin; |
||||
} |
||||
|
||||
totalWeight += weight; |
||||
} |
||||
|
||||
|
||||
if (!updateLock.get() && hosts.size() != map.size() && updateLock.compareAndSet(false, true)) { |
||||
try { |
||||
ConcurrentMap<String, WeightedRoundRobin> newMap = new ConcurrentHashMap<>(map); |
||||
newMap.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD); |
||||
workGroupWeightMap.put(key, newMap); |
||||
} finally { |
||||
updateLock.set(false); |
||||
} |
||||
} |
||||
|
||||
if (selectedHost != null) { |
||||
selectWeightRoundRobin.sel(totalWeight); |
||||
return selectedHost; |
||||
} |
||||
|
||||
return hosts.get(0); |
||||
} |
||||
} |
||||
@ -1,29 +0,0 @@
|
||||
package org.apache.dolphinscheduler.remote.rpc.selector; |
||||
|
||||
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); |
||||
|
||||
} |
||||
@ -1,44 +0,0 @@
|
||||
package org.apache.dolphinscheduler.remote.rpc.selector; |
||||
|
||||
import org.apache.dolphinscheduler.remote.utils.Host; |
||||
|
||||
import java.util.ArrayList; |
||||
import java.util.Collection; |
||||
import java.util.List; |
||||
import java.util.concurrent.ThreadLocalRandom; |
||||
|
||||
/** |
||||
* RandomSelector |
||||
*/ |
||||
public class RandomSelector extends AbstractSelector<Host> { |
||||
|
||||
@Override |
||||
public Host doSelect(final Collection<Host> source) { |
||||
|
||||
List<Host> hosts = new ArrayList<>(source); |
||||
int size = hosts.size(); |
||||
int[] weights = new int[size]; |
||||
int totalWeight = 0; |
||||
int index = 0; |
||||
|
||||
for (Host host : hosts) { |
||||
totalWeight += host.getWeight(); |
||||
weights[index] = host.getWeight(); |
||||
index++; |
||||
} |
||||
|
||||
if (totalWeight > 0) { |
||||
int offset = ThreadLocalRandom.current().nextInt(totalWeight); |
||||
|
||||
for (int i = 0; i < size; i++) { |
||||
offset -= weights[i]; |
||||
if (offset < 0) { |
||||
return hosts.get(i); |
||||
} |
||||
} |
||||
} |
||||
return hosts.get(ThreadLocalRandom.current().nextInt(size)); |
||||
} |
||||
|
||||
} |
||||
|
||||
@ -1,16 +0,0 @@
|
||||
package org.apache.dolphinscheduler.remote.rpc.selector; |
||||
|
||||
import java.util.Collection; |
||||
|
||||
/** |
||||
* Selector |
||||
*/ |
||||
public interface Selector<T> { |
||||
|
||||
/** |
||||
* select |
||||
* @param source source |
||||
* @return T |
||||
*/ |
||||
T select(Collection<T> source); |
||||
} |
||||
Loading…
Reference in new issue