natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Adaptive loadbalance (#10745)

This commit is contained in:
Starfish-Ning 2022-12-13 16:32:49 +08:00 committed by GitHub
parent 029b15fa16
commit d17160591c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 604 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/ClusterScopeModelInitializer.java
View File

@ -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

133
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/AdaptiveLoadBalance.java Normal file
View File

@ -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
* </p>
*/
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 <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
Invoker<T> 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 <T> Invoker<T> selectByP2C(List<Invoker<T>> 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 <T> Invoker<T> chooseLowLoadInvoker(Invoker<T> invoker1,Invoker<T> 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;
}
}

3
dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.LoadBalance
View File

@ -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
shortestresponse=org.apache.dubbo.rpc.cluster.loadbalance.ShortestResponseLoadBalance
adaptive=org.apache.dubbo.rpc.cluster.loadbalance.AdaptiveLoadBalance

167
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/AdaptiveLoadBalanceTest.java Normal file
View File

@ -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<String, String> 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<Invoker<LoadBalanceBaseTest>, 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!");
}
}

5
dubbo-common/src/main/java/org/apache/dubbo/common/constants/LoadbalanceRules.java
View File

@ -46,6 +46,11 @@ public interface LoadbalanceRules {
**/
String SHORTEST_RESPONSE = "shortestresponse";
/**
* adaptive load balance.
**/
String ADAPTIVE = "adaptive";
String EMPTY = "";
}

121
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AdaptiveMetrics.java Normal file
View File

@ -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<String, AdaptiveMetrics> 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<String,String> 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;
}
}

3
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java
View File

@ -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";
}

151
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/AdaptiveLoadBalanceFilter.java Normal file
View File

@ -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<String, String> 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));
});
}
}
}

19
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ProfilerServerFilter.java
View File

@ -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;

1
dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter
View File

@ -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