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

Support consistent hash load balance when gerenic invoke (#10799)

This commit is contained in:
nobodyw-cell 2022-11-06 17:13:52 +08:00 committed by GitHub
parent b319fce427
commit a204f61839
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 60 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalance.java
View File

@ -28,6 +28,7 @@ import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.constants.CommonConstants.$INVOKE;
import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
/**
@ -96,15 +97,21 @@ public class ConsistentHashLoadBalance extends AbstractLoadBalance {
}
public Invoker<T> select(Invocation invocation) {
String key = toKey(invocation.getArguments());
boolean isGeneric = invocation.getMethodName().equals($INVOKE);
String key = toKey(invocation.getArguments(),isGeneric);
byte[] digest = Bytes.getMD5(key);
return selectForKey(hash(digest, 0));
}
private String toKey(Object[] args, boolean isGeneric) {
return isGeneric ? toKey((Object[]) args[1]) : toKey(args);
}
private String toKey(Object[] args) {
StringBuilder buf = new StringBuilder();
for (int i : argumentIndex) {
if (i >= 0 && i < args.length) {
if (i >= 0 && args != null && i < args.length) {
buf.append(args[i]);
}
}

28
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalanceTest.java
View File

@ -33,6 +33,34 @@ import java.util.concurrent.atomic.AtomicLong;
@SuppressWarnings("rawtypes")
public class ConsistentHashLoadBalanceTest extends LoadBalanceBaseTest {
@Test
public void testConsistentHashLoadBalanceInGenericCall() {
int runs = 10000;
Map<Invoker, AtomicLong> genericInvokeCounter = getGenericInvokeCounter(runs, ConsistentHashLoadBalance.NAME);
Map<Invoker, AtomicLong> invokeCounter = getInvokeCounter(runs, ConsistentHashLoadBalance.NAME);
Invoker genericHitted = findHitted(genericInvokeCounter);
Invoker hitted = findHitted(invokeCounter);
Assertions.assertEquals(hitted,
genericHitted, "hitted should equals to genericHitted");
}
private Invoker findHitted(Map<Invoker,AtomicLong> invokerCounter) {
Invoker invoker = null;
for (Map.Entry<Invoker,AtomicLong> entry : invokerCounter.entrySet()) {
if (entry.getValue().longValue() > 0) {
invoker = entry.getKey();
break;
}
}
Assertions.assertNotNull(invoker,"invoker should be found");
return null;
}
@Test
public void testConsistentHashLoadBalance() {
int runs = 10000;

23
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/LoadBalanceBaseTest.java
View File

@ -48,6 +48,7 @@ import static org.mockito.Mockito.mock;
@SuppressWarnings({"unchecked", "rawtypes"})
public class LoadBalanceBaseTest {
Invocation invocation;
Invocation genericInvocation;
List<Invoker<LoadBalanceBaseTest>> invokers = new ArrayList<Invoker<LoadBalanceBaseTest>>();
Invoker<LoadBalanceBaseTest> invoker1;
Invoker<LoadBalanceBaseTest> invoker2;
@ -79,6 +80,14 @@ public class LoadBalanceBaseTest {
given(invocation.getMethodName()).willReturn("method1");
given(invocation.getArguments()).willReturn(new Object[] {"arg1","arg2","arg3"});
genericInvocation = mock(Invocation.class);
String methodName = "method1";
given(genericInvocation.getMethodName()).willReturn("$invoke");
String[] paraTypes = new String[] {String.class.getName(),String.class.getName(),String.class.getName()};
Object[] argsObject = new Object[] {"arg1","arg2","arg3"};
Object[] args = new Object[] {methodName,paraTypes,argsObject};
given(genericInvocation.getArguments()).willReturn(args);
invoker1 = mock(Invoker.class);
invoker2 = mock(Invoker.class);
invoker3 = mock(Invoker.class);
@ -132,6 +141,20 @@ public class LoadBalanceBaseTest {
return counter;
}
public Map<Invoker, AtomicLong> getGenericInvokeCounter(int runs, String loadbalanceName) {
Map<Invoker, AtomicLong> counter = new ConcurrentHashMap<Invoker, AtomicLong>();
LoadBalance lb = getLoadBalance(loadbalanceName);
for (Invoker invoker : invokers) {
counter.put(invoker, new AtomicLong(0));
}
URL url = invokers.get(0).getUrl();
for (int i = 0; i < runs; i++) {
Invoker sinvoker = lb.select(invokers, url, genericInvocation);
counter.get(sinvoker).incrementAndGet();
}
return counter;
}
protected AbstractLoadBalance getLoadBalance(String loadbalanceName) {
return (AbstractLoadBalance) ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(loadbalanceName);
}