diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/ClusterScopeModelInitializer.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/ClusterScopeModelInitializer.java index 4b2e01b71d..c7a434b421 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/ClusterScopeModelInitializer.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/ClusterScopeModelInitializer.java @@ -17,6 +17,7 @@ package org.apache.dubbo.rpc.cluster; import org.apache.dubbo.common.beans.factory.ScopeBeanFactory; +import org.apache.dubbo.rpc.AdaptiveMetrics; import org.apache.dubbo.rpc.cluster.merger.MergerFactory; import org.apache.dubbo.rpc.cluster.router.RouterSnapshotSwitcher; import org.apache.dubbo.rpc.cluster.router.mesh.route.MeshRuleManager; @@ -38,6 +39,7 @@ public class ClusterScopeModelInitializer implements ScopeModelInitializer { ScopeBeanFactory beanFactory = applicationModel.getBeanFactory(); beanFactory.registerBean(MergerFactory.class); beanFactory.registerBean(ClusterUtils.class); + beanFactory.registerBean(AdaptiveMetrics.class); } @Override diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/AdaptiveLoadBalance.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/AdaptiveLoadBalance.java new file mode 100644 index 0000000000..8bf9648300 --- /dev/null +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/AdaptiveLoadBalance.java @@ -0,0 +1,133 @@ +/* + * 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.dubbo.rpc.cluster.loadbalance; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.constants.LoadbalanceRules; +import org.apache.dubbo.common.utils.StringUtils; +import org.apache.dubbo.rpc.AdaptiveMetrics; +import org.apache.dubbo.rpc.Constants; +import org.apache.dubbo.rpc.Invocation; +import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.RpcContext; +import org.apache.dubbo.rpc.model.ApplicationModel; +import org.apache.dubbo.rpc.support.RpcUtils; + +import java.util.List; +import java.util.concurrent.ThreadLocalRandom; + +import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_TIMEOUT; +import static org.apache.dubbo.common.constants.CommonConstants.LOADBALANCE_KEY; + +/** + * AdaptiveLoadBalance + *

+ */ +public class AdaptiveLoadBalance extends AbstractLoadBalance { + + public static final String NAME = "adaptive"; + + //default key + private String attachmentKey = "mem,load"; + + private AdaptiveMetrics adaptiveMetrics; + + public AdaptiveLoadBalance(ApplicationModel scopeModel){ + adaptiveMetrics = scopeModel.getBeanFactory().getBean(AdaptiveMetrics.class); + } + + @Override + protected Invoker doSelect(List> invokers, URL url, Invocation invocation) { + Invoker invoker = selectByP2C(invokers,invocation); + invocation.setAttachment(Constants.ADAPTIVE_LOADBALANCE_ATTACHMENT_KEY,attachmentKey); + long startTime = System.currentTimeMillis(); + invocation.getAttributes().put(Constants.ADAPTIVE_LOADBALANCE_START_TIME,startTime); + invocation.getAttributes().put(LOADBALANCE_KEY,LoadbalanceRules.ADAPTIVE); + adaptiveMetrics.addConsumerReq(getServiceKey(invoker,invocation)); + adaptiveMetrics.setPickTime(getServiceKey(invoker,invocation),startTime); + + return invoker; + } + + private Invoker selectByP2C(List> invokers, Invocation invocation){ + int length = invokers.size(); + if(length == 1) { + return invokers.get(0); + } + + if(length == 2) { + return chooseLowLoadInvoker(invokers.get(0),invokers.get(1),invocation); + } + + int pos1 = ThreadLocalRandom.current().nextInt(length); + int pos2 = ThreadLocalRandom.current().nextInt(length - 1); + if (pos2 >= pos1) { + pos2 = pos2 + 1; + } + + return chooseLowLoadInvoker(invokers.get(pos1),invokers.get(pos2),invocation); + } + + private String getServiceKey(Invoker invoker,Invocation invocation){ + + String key = (String) invocation.getAttributes().get(invoker); + if (StringUtils.isNotEmpty(key)){ + return key; + } + + key = buildServiceKey(invoker,invocation); + invocation.getAttributes().put(invoker,key); + return key; + } + + private String buildServiceKey(Invoker invoker,Invocation invocation){ + URL url = invoker.getUrl(); + StringBuilder sb = new StringBuilder(128); + sb.append(url.getAddress()).append(":").append(invocation.getProtocolServiceKey()); + return sb.toString(); + } + + private int getTimeout(Invoker invoker, Invocation invocation) { + URL url = invoker.getUrl(); + String methodName = RpcUtils.getMethodName(invocation); + return (int) RpcUtils.getTimeout(url,methodName, RpcContext.getClientAttachment(),invocation, DEFAULT_TIMEOUT); + } + + private Invoker chooseLowLoadInvoker(Invoker invoker1,Invoker invoker2,Invocation invocation){ + int weight1 = getWeight(invoker1, invocation); + int weight2 = getWeight(invoker2, invocation); + int timeout1 = getTimeout(invoker2, invocation); + int timeout2 = getTimeout(invoker2, invocation); + long load1 = Double.doubleToLongBits(adaptiveMetrics.getLoad(getServiceKey(invoker1,invocation),weight1,timeout1 )); + long load2 = Double.doubleToLongBits(adaptiveMetrics.getLoad(getServiceKey(invoker2,invocation),weight2,timeout2 )); + + if (load1 == load2) { + // The sum of weights + int totalWeight = weight1 + weight2; + if (totalWeight > 0) { + int offset = ThreadLocalRandom.current().nextInt(totalWeight); + if (offset < weight1) { + return invoker1; + } + return invoker2; + } + return ThreadLocalRandom.current().nextInt(2) == 0 ? invoker1 : invoker2; + } + return load1 > load2 ? invoker2 : invoker1; + } + +} diff --git a/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.LoadBalance b/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.LoadBalance index 347ea10b57..1bb371b7f7 100644 --- a/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.LoadBalance +++ b/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.LoadBalance @@ -2,4 +2,5 @@ random=org.apache.dubbo.rpc.cluster.loadbalance.RandomLoadBalance roundrobin=org.apache.dubbo.rpc.cluster.loadbalance.RoundRobinLoadBalance leastactive=org.apache.dubbo.rpc.cluster.loadbalance.LeastActiveLoadBalance consistenthash=org.apache.dubbo.rpc.cluster.loadbalance.ConsistentHashLoadBalance -shortestresponse=org.apache.dubbo.rpc.cluster.loadbalance.ShortestResponseLoadBalance \ No newline at end of file +shortestresponse=org.apache.dubbo.rpc.cluster.loadbalance.ShortestResponseLoadBalance +adaptive=org.apache.dubbo.rpc.cluster.loadbalance.AdaptiveLoadBalance diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/AdaptiveLoadBalanceTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/AdaptiveLoadBalanceTest.java new file mode 100644 index 0000000000..1a8e90f5ff --- /dev/null +++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/AdaptiveLoadBalanceTest.java @@ -0,0 +1,167 @@ +/* + * 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.dubbo.rpc.cluster.loadbalance; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.rpc.AdaptiveMetrics; +import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.model.ApplicationModel; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.MethodOrderer; +import org.junit.jupiter.api.Order; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestMethodOrder; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; +import java.util.stream.Collectors; + +@TestMethodOrder(MethodOrderer.OrderAnnotation.class) +class AdaptiveLoadBalanceTest extends LoadBalanceBaseTest { + + private ApplicationModel scopeModel; + + private AdaptiveMetrics adaptiveMetrics; + + @Test + @Order(0) + void testSelectByWeight() { + int sumInvoker1 = 0; + int sumInvoker2 = 0; + int sumInvoker3 = 0; + int loop = 10000; + + ApplicationModel scopeModel = ApplicationModel.defaultModel(); + + AdaptiveLoadBalance lb = new AdaptiveLoadBalance(scopeModel); + for (int i = 0; i < loop; i++) { + Invoker selected = lb.select(weightInvokers, null, weightTestInvocation); + + if (selected.getUrl().getProtocol().equals("test1")) { + sumInvoker1++; + } + + if (selected.getUrl().getProtocol().equals("test2")) { + sumInvoker2++; + } + + if (selected.getUrl().getProtocol().equals("test3")) { + sumInvoker3++; + } + } + + // 1 : 9 : 6 + System.out.println(sumInvoker1); + System.out.println(sumInvoker2); + System.out.println(sumInvoker3); + Assertions.assertEquals(sumInvoker1 + sumInvoker2 + sumInvoker3, loop, "select failed!"); + } + + private String buildServiceKey(Invoker invoker){ + URL url = invoker.getUrl(); + return url.getAddress() + ":" + invocation.getProtocolServiceKey(); + } + + private AdaptiveMetrics getAdaptiveMetricsInstance(){ + if (adaptiveMetrics == null) { + adaptiveMetrics = scopeModel.getBeanFactory().getBean(AdaptiveMetrics.class); + } + return adaptiveMetrics; + } + + @Test + @Order(1) + void testSelectByAdaptive() { + int sumInvoker1 = 0; + int sumInvoker2 = 0; + int sumInvoker5 = 0; + int loop = 10000; + + scopeModel = ApplicationModel.defaultModel(); + AdaptiveLoadBalance lb = new AdaptiveLoadBalance(scopeModel); + + lb.select(weightInvokersSR, null, weightTestInvocation); + + for (int i = 0; i < loop; i++) { + Invoker selected = lb.select(weightInvokersSR, null, weightTestInvocation); + + Map metricsMap = new HashMap<>(); + String idKey = buildServiceKey(selected); + + if (selected.getUrl().getProtocol().equals("test1")) { + sumInvoker1++; + metricsMap.put("rt", "10"); + metricsMap.put("load", "10"); + metricsMap.put("curTime", String.valueOf(System.currentTimeMillis()-10)); + getAdaptiveMetricsInstance().addConsumerSuccess(idKey); + } + + if (selected.getUrl().getProtocol().equals("test2")) { + sumInvoker2++; + metricsMap.put("rt", "100"); + metricsMap.put("load", "40"); + metricsMap.put("curTime", String.valueOf(System.currentTimeMillis()-100)); + getAdaptiveMetricsInstance().addConsumerSuccess(idKey); + } + + if (selected.getUrl().getProtocol().equals("test5")) { + metricsMap.put("rt", "5000"); + metricsMap.put("load", "400");//400% + metricsMap.put("curTime", String.valueOf(System.currentTimeMillis() - 5000)); + + getAdaptiveMetricsInstance().addErrorReq(idKey); + sumInvoker5++; + } + getAdaptiveMetricsInstance().setProviderMetrics(idKey,metricsMap); + + } + Map, Integer> weightMap = weightInvokersSR.stream() + .collect(Collectors.toMap(Function.identity(), e -> Integer.valueOf(e.getUrl().getParameter("weight")))); + Integer totalWeight = weightMap.values().stream().reduce(0, Integer::sum); + // max deviation = expectWeightValue * 2 + int expectWeightValue = loop / totalWeight; + int maxDeviation = expectWeightValue * 2; + double beta = 0.5; + //this EMA is an approximate value + double ewma1 = beta * 50 + (1 - beta) * 10; + double ewma2 = beta * 50 + (1 - beta) * 100; + double ewma5 = beta * 50 + (1 - beta) * 5000; + + AtomicInteger weight1 = new AtomicInteger(); + AtomicInteger weight2 = new AtomicInteger(); + AtomicInteger weight5 = new AtomicInteger(); + weightMap.forEach((k, v) ->{ + if (k.getUrl().getProtocol().equals("test1")){ + weight1.set(v); + } + else if (k.getUrl().getProtocol().equals("test2")){ + weight2.set(v); + } + else if (k.getUrl().getProtocol().equals("test5")){ + weight5.set(v); + } + }); + + Assertions.assertEquals(sumInvoker1 + sumInvoker2 + sumInvoker5, loop, "select failed!"); + Assertions.assertTrue(Math.abs(sumInvoker1 / (weightMap.get(weightInvoker1) * ewma1) - expectWeightValue) < maxDeviation, "select failed!"); + Assertions.assertTrue(Math.abs(sumInvoker2 / (weightMap.get(weightInvoker2) * ewma2) - expectWeightValue) < maxDeviation, "select failed!"); + Assertions.assertTrue(Math.abs(sumInvoker5 / (weightMap.get(weightInvoker5) * ewma5) - expectWeightValue) < maxDeviation, "select failed!"); + } +} diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/LoadbalanceRules.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/LoadbalanceRules.java index daeff6c1bd..b129d9d4e4 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/LoadbalanceRules.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/LoadbalanceRules.java @@ -46,6 +46,11 @@ public interface LoadbalanceRules { **/ String SHORTEST_RESPONSE = "shortestresponse"; + /** + * adaptive load balance. + **/ + String ADAPTIVE = "adaptive"; + String EMPTY = ""; } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AdaptiveMetrics.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AdaptiveMetrics.java new file mode 100644 index 0000000000..0c6fa4cce6 --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AdaptiveMetrics.java @@ -0,0 +1,121 @@ +/* + * 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.dubbo.rpc; + +import org.apache.dubbo.common.utils.StringUtils; + +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicLong; + +/** + * adaptive Metrics statistics. + */ +public class AdaptiveMetrics { + + private ConcurrentMap metricsStatistics = new ConcurrentHashMap<>(); + + private long currentProviderTime = 0; + private double providerCPULoad = 0; + private long lastLatency = 0; + private long currentTime = 0; + + //Allow some time disorder + private long pickTime = System.currentTimeMillis(); + + private double beta = 0.5; + private final AtomicLong consumerReq = new AtomicLong(); + private final AtomicLong consumerSuccess = new AtomicLong(); + private final AtomicLong errorReq = new AtomicLong(); + private double ewma = 0; + + public double getLoad(String idKey,int weight,int timeout){ + AdaptiveMetrics metrics = getStatus(idKey); + + //If the time more than 2 times, mandatory selected + if (System.currentTimeMillis() - metrics.pickTime > timeout * 2) { + return 0; + } + + if (metrics.currentTime > 0){ + long multiple = (System.currentTimeMillis() - metrics.currentTime) / timeout + 1; + if (multiple > 0) { + if (metrics.currentProviderTime == metrics.currentTime) { + //penalty value + metrics.lastLatency = timeout * 2L; + }else { + metrics.lastLatency = metrics.lastLatency >> multiple; + } + metrics.ewma = metrics.beta * metrics.ewma + (1 - metrics.beta) * metrics.lastLatency; + metrics.currentTime = System.currentTimeMillis(); + } + } + + long inflight = metrics.consumerReq.get() - metrics.consumerSuccess.get() - metrics.errorReq.get(); + return metrics.providerCPULoad * (Math.sqrt(metrics.ewma) + 1) * (inflight + 1) / ((((double)metrics.consumerSuccess.get() / (double)(metrics.consumerReq.get() + 1)) * weight) + 1); + } + + public AdaptiveMetrics getStatus(String idKey){ + return metricsStatistics.computeIfAbsent(idKey, k -> new AdaptiveMetrics()); + } + + public void addConsumerReq(String idKey){ + AdaptiveMetrics metrics = getStatus(idKey); + metrics.consumerReq.incrementAndGet(); + } + + public void addConsumerSuccess(String idKey){ + AdaptiveMetrics metrics = getStatus(idKey); + metrics.consumerSuccess.incrementAndGet(); + } + + public void addErrorReq(String idKey){ + AdaptiveMetrics metrics = getStatus(idKey); + metrics.errorReq.incrementAndGet(); + } + + public void setPickTime(String idKey,long time){ + AdaptiveMetrics metrics = getStatus(idKey); + metrics.pickTime = time; + } + + + + public void setProviderMetrics(String idKey,Map metricsMap){ + + AdaptiveMetrics metrics = getStatus(idKey); + + long serviceTime = Long.parseLong(Optional.ofNullable(metricsMap.get("curTime")).filter(v -> StringUtils.isNumeric(v,false)).orElse("0")); + //If server time is less than the current time, discard + if (metrics.currentProviderTime > serviceTime){ + return; + } + + metrics.currentProviderTime = serviceTime; + metrics.currentTime = serviceTime; + metrics.providerCPULoad = Double.parseDouble(Optional.ofNullable(metricsMap.get("load")).filter(v -> StringUtils.isNumeric(v,true)).orElse("0")); + metrics.lastLatency = Long.parseLong((Optional.ofNullable(metricsMap.get("rt")).filter(v -> StringUtils.isNumeric(v,false)).orElse("0"))); + + metrics.beta = 0.5; + //Vt = β * Vt-1 + (1 - β ) * θt + metrics.ewma = metrics.beta * metrics.ewma + (1 - metrics.beta) * metrics.lastLatency; + + } +} + diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java index eb220cf752..f235bd83d3 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java @@ -104,9 +104,12 @@ public interface Constants { String H2_SETTINGS_MAX_FRAME_SIZE_KEY = "dubbo.rpc.tri.max-frame-size"; String H2_SETTINGS_MAX_HEADER_LIST_SIZE_KEY = "dubbo.rpc.tri.max-header-list-size"; + String ADAPTIVE_LOADBALANCE_ATTACHMENT_KEY = "lb_adaptive"; + String ADAPTIVE_LOADBALANCE_START_TIME = "adaptive_startTime"; String H2_SUPPORT_NO_LOWER_HEADER_KEY = "dubbo.rpc.tri.support-no-lower-header"; String H2_IGNORE_1_0_0_KEY = "dubbo.rpc.tri.ignore-1.0.0-version"; String H2_RESOLVE_FALLBACK_TO_DEFAULT_KEY = "dubbo.rpc.tri.resolve-fallback-to-default"; + } diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/AdaptiveLoadBalanceFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/AdaptiveLoadBalanceFilter.java new file mode 100644 index 0000000000..0ec2327cbc --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/AdaptiveLoadBalanceFilter.java @@ -0,0 +1,151 @@ +/* + * 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.dubbo.rpc.filter; + +import org.apache.dubbo.common.constants.LoadbalanceRules; +import org.apache.dubbo.common.extension.Activate; +import org.apache.dubbo.common.resource.GlobalResourcesRepository; +import org.apache.dubbo.common.threadlocal.NamedInternalThreadFactory; +import org.apache.dubbo.common.utils.StringUtils; +import org.apache.dubbo.rpc.AdaptiveMetrics; +import org.apache.dubbo.rpc.Constants; +import org.apache.dubbo.rpc.Filter; +import org.apache.dubbo.rpc.Invocation; +import org.apache.dubbo.rpc.Invoker; +import org.apache.dubbo.rpc.Result; +import org.apache.dubbo.rpc.RpcException; +import org.apache.dubbo.rpc.model.ApplicationModel; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN; +import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER; +import static org.apache.dubbo.common.constants.CommonConstants.LOADBALANCE_KEY; + +/** + * if the load balance is adaptive ,set attachment to get the metrics of the server + * @see org.apache.dubbo.rpc.Filter + * @see org.apache.dubbo.rpc.RpcContext + */ +@Activate(group = CONSUMER, order = -200000, value = {"loadbalance:adaptive"}) +public class AdaptiveLoadBalanceFilter implements Filter, Filter.Listener { + + /** + * uses a single worker thread operating off an bounded queue + */ + private volatile ThreadPoolExecutor executor = null; + + private AdaptiveMetrics adaptiveMetrics; + + public AdaptiveLoadBalanceFilter(ApplicationModel scopeModel) { + adaptiveMetrics = scopeModel.getBeanFactory().getBean(AdaptiveMetrics.class); + } + + private ThreadPoolExecutor getExecutor(){ + if (null == executor) { + synchronized (this) { + if (null == executor) { + executor = new ThreadPoolExecutor(1, 1, 0L,TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), + new NamedInternalThreadFactory("Dubbo-framework-loadbalance-adaptive", true), new ThreadPoolExecutor.DiscardOldestPolicy()); + GlobalResourcesRepository.getInstance().registerDisposable(() -> this.executor.shutdown()); + } + } + } + return executor; + } + + @Override + public Result invoke(Invoker invoker, Invocation invocation) throws RpcException { + return invoker.invoke(invocation); + } + + private String buildServiceKey(Invocation invocation){ + StringBuilder sb = new StringBuilder(128); + sb.append(invocation.getInvoker().getUrl().getAddress()).append(":").append(invocation.getProtocolServiceKey()); + return sb.toString(); + } + + private String getServiceKey(Invocation invocation){ + + String key = (String) invocation.getAttributes().get(invocation.getInvoker()); + if (StringUtils.isNotEmpty(key)){ + return key; + } + + key = buildServiceKey(invocation); + invocation.getAttributes().put(invocation.getInvoker(),key); + return key; + } + + @Override + public void onResponse(Result appResponse, Invoker invoker, Invocation invocation) { + + try { + String loadBalance = (String) invocation.getAttributes().get(LOADBALANCE_KEY); + if (StringUtils.isEmpty(loadBalance) + || !LoadbalanceRules.ADAPTIVE.equals(loadBalance)) { + return; + } + adaptiveMetrics.addConsumerSuccess(getServiceKey(invocation)); + String attachment = appResponse.getAttachment(Constants.ADAPTIVE_LOADBALANCE_ATTACHMENT_KEY); + if (StringUtils.isNotEmpty(attachment)) { + String[] parties = COMMA_SPLIT_PATTERN.split(attachment); + if (parties.length == 0) { + return; + } + Map metricsMap = new HashMap<>(); + for (String party : parties) { + String[] groups = party.split(":"); + if (groups.length != 2) { + continue; + } + metricsMap.put(groups[0], groups[1]); + } + + Long startTime = (Long) invocation.getAttributes().get(Constants.ADAPTIVE_LOADBALANCE_START_TIME); + if (null != startTime) { + metricsMap.put("rt", String.valueOf(System.currentTimeMillis() - startTime)); + } + + getExecutor().execute(() -> { + adaptiveMetrics.setProviderMetrics(getServiceKey(invocation), metricsMap); + }); + } + } + finally { + appResponse.getAttachments().remove(Constants.ADAPTIVE_LOADBALANCE_ATTACHMENT_KEY); + } + + } + + @Override + public void onError(Throwable t, Invoker invoker, Invocation invocation) { + String loadBalance = (String) invocation.getAttributes().get(LOADBALANCE_KEY); + if (StringUtils.isNotEmpty(loadBalance) + && LoadbalanceRules.ADAPTIVE.equals(loadBalance)) { + getExecutor().execute(() -> { + adaptiveMetrics.addErrorReq(getServiceKey(invocation)); + }); + } + } + + +} diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ProfilerServerFilter.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ProfilerServerFilter.java index 1fb19f2dde..e1b1c9dc43 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ProfilerServerFilter.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ProfilerServerFilter.java @@ -22,6 +22,7 @@ import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.profiler.Profiler; import org.apache.dubbo.common.profiler.ProfilerEntry; import org.apache.dubbo.common.profiler.ProfilerSwitch; +import org.apache.dubbo.common.utils.StringUtils; import org.apache.dubbo.rpc.BaseFilter; import org.apache.dubbo.rpc.Filter; import org.apache.dubbo.rpc.Invocation; @@ -29,10 +30,15 @@ import org.apache.dubbo.rpc.Invoker; import org.apache.dubbo.rpc.Result; import org.apache.dubbo.rpc.RpcContext; import org.apache.dubbo.rpc.RpcException; +import org.apache.dubbo.rpc.Constants; + +import java.lang.management.ManagementFactory; +import java.lang.management.OperatingSystemMXBean; import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_TIMEOUT; import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER; import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SEPARATOR; import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROXY_TIMEOUT_RESPONSE; @Activate(group = PROVIDER, order = Integer.MIN_VALUE) @@ -60,6 +66,7 @@ public class ProfilerServerFilter implements Filter, BaseFilter.Listener { @Override public void onResponse(Result appResponse, Invoker invoker, Invocation invocation) { afterInvoke(invoker, invocation); + addAdaptiveResponse(appResponse, invocation); } @Override @@ -78,6 +85,18 @@ public class ProfilerServerFilter implements Filter, BaseFilter.Listener { } } } + private void addAdaptiveResponse(Result appResponse, Invocation invocation) { + String adaptiveLoadAttachment = invocation.getAttachment(Constants.ADAPTIVE_LOADBALANCE_ATTACHMENT_KEY); + if (StringUtils.isNotEmpty(adaptiveLoadAttachment)) { + OperatingSystemMXBean operatingSystemMXBean = ManagementFactory.getOperatingSystemMXBean(); + + StringBuilder sb = new StringBuilder(64); + sb.append("curTime:").append(System.currentTimeMillis()); + sb.append(COMMA_SEPARATOR).append("load:").append(operatingSystemMXBean.getSystemLoadAverage() * 100 / operatingSystemMXBean.getAvailableProcessors() ); + + appResponse.setAttachment(Constants.ADAPTIVE_LOADBALANCE_ATTACHMENT_KEY,sb.toString()); + } + } private void dumpIfNeed(Invoker invoker, Invocation invocation, ProfilerEntry profiler) { int timeout; diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter b/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter index bed37fabab..792fbe4f30 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter +++ b/dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter @@ -13,3 +13,4 @@ compatible=org.apache.dubbo.rpc.filter.CompatibleFilter timeout=org.apache.dubbo.rpc.filter.TimeoutFilter tps=org.apache.dubbo.rpc.filter.TpsLimitFilter profiler-server=org.apache.dubbo.rpc.filter.ProfilerServerFilter +adaptiveLoadBalance=org.apache.dubbo.rpc.filter.AdaptiveLoadBalanceFilter