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

Distinguish getUrl and getConsumerUrl from Directory. (#5775) 

* distinguish getUrl and getConsumerUrl from Directory
This commit is contained in:
ken.lj 2020-02-27 10:43:18 +08:00 committed by GitHub
parent 0f4e60dca3
commit 048fadf2c0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 34 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java
View File

@ -85,11 +85,11 @@ public abstract class AbstractClusterInvoker<T> implements Invoker<T> {
@Override
public URL getUrl() {
return directory.getUrl();
return directory.getConsumerUrl();
}
protected URL getConsumerUrl() {
return directory.getConsumerUrl();
public URL getRegistryUrl() {
return directory.getUrl();
}
@Override
@ -255,7 +255,7 @@ public abstract class AbstractClusterInvoker<T> implements Invoker<T> {
List<Invoker<T>> invokers = list(invocation);
LoadBalance loadbalance = initLoadBalance(invokers, invocation);
RpcUtils.attachInvocationIdIfAsync(getConsumerUrl(), invocation);
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
return doInvoke(invocation, invokers, loadbalance);
}

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

@ -58,7 +58,7 @@ public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {
List<Invoker<T>> copyInvokers = invokers;
checkInvokers(copyInvokers, invocation);
String methodName = RpcUtils.getMethodName(invocation);
int len = getConsumerUrl().getMethodParameter(methodName, RETRIES_KEY, DEFAULT_RETRIES) + 1;
int len = getUrl().getMethodParameter(methodName, RETRIES_KEY, DEFAULT_RETRIES) + 1;
if (len <= 0) {
len = 1;
}

4
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvoker.java
View File

@ -65,8 +65,8 @@ public class ForkingClusterInvoker<T> extends AbstractClusterInvoker<T> {
try {
checkInvokers(invokers, invocation);
final List<Invoker<T>> selected;
final int forks = getConsumerUrl().getParameter(FORKS_KEY, DEFAULT_FORKS);
final int timeout = getConsumerUrl().getParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT);
final int forks = getUrl().getParameter(FORKS_KEY, DEFAULT_FORKS);
final int timeout = getUrl().getParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT);
if (forks <= 0 || forks >= invokers.size()) {
selected = invokers;
} else {

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

@ -63,7 +63,7 @@ public class MergeableClusterInvoker<T> extends AbstractClusterInvoker<T> {
@Override
protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
checkInvokers(invokers, invocation);
String merger = getConsumerUrl().getMethodParameter(invocation.getMethodName(), MERGER_KEY);
String merger = getUrl().getMethodParameter(invocation.getMethodName(), MERGER_KEY);
if (ConfigUtils.isEmpty(merger)) { // If a method doesn't have a merger, only invoke one Group
for (final Invoker<T> invoker : invokers) {
if (invoker.isAvailable()) {

22
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/registry/ZoneAwareClusterInvoker.java
View File

@ -26,6 +26,7 @@ import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.Directory;
import org.apache.dubbo.rpc.cluster.LoadBalance;
import org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker;
import org.apache.dubbo.rpc.cluster.support.wrapper.MockClusterInvoker;
import java.util.List;
import java.util.stream.Collectors;
@ -58,24 +59,28 @@ public class ZoneAwareClusterInvoker<T> extends AbstractClusterInvoker<T> {
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
// First, pick the invoker (XXXClusterInvoker) that comes from the local registry, distinguish by a 'preferred' key.
for (Invoker<T> invoker : invokers) {
if (invoker.isAvailable() && invoker.getUrl().getParameter(REGISTRY_KEY + "." + PREFERRED_KEY, false)) {
return invoker.invoke(invocation);
// FIXME, the invoker is a cluster invoker representing one Registry, so it will automatically wrapped by MockClusterInvoker.
MockClusterInvoker<T> mockClusterInvoker = (MockClusterInvoker<T>) invoker;
if (mockClusterInvoker.isAvailable() && mockClusterInvoker.getRegistryUrl()
.getParameter(REGISTRY_KEY + "." + PREFERRED_KEY, false)) {
return mockClusterInvoker.invoke(invocation);
}
}
// providers in the registry with the same
// providers in the registry with the same zone
String zone = (String) invocation.getAttachment(REGISTRY_ZONE);
if (StringUtils.isNotEmpty(zone)) {
for (Invoker<T> invoker : invokers) {
if (invoker.isAvailable() && zone.equals(invoker.getUrl().getParameter(REGISTRY_KEY + "." + ZONE_KEY))) {
return invoker.invoke(invocation);
MockClusterInvoker<T> mockClusterInvoker = (MockClusterInvoker<T>) invoker;
if (mockClusterInvoker.isAvailable() && zone.equals(mockClusterInvoker.getRegistryUrl().getParameter(REGISTRY_KEY + "." + ZONE_KEY))) {
return mockClusterInvoker.invoke(invocation);
}
}
String force = (String) invocation.getAttachment(REGISTRY_ZONE_FORCE);
if (StringUtils.isNotEmpty(force) && "true".equalsIgnoreCase(force)) {
throw new IllegalStateException("No registry instance in zone or no available providers in the registry, zone: "
+ zone
+ ", registries: " + invokers.stream().map(i -> i.getUrl().toString()).collect(Collectors.joining(",")));
+ ", registries: " + invokers.stream().map(invoker -> ((MockClusterInvoker<T>) invoker).getRegistryUrl().toString()).collect(Collectors.joining(",")));
}
}
@ -88,8 +93,9 @@ public class ZoneAwareClusterInvoker<T> extends AbstractClusterInvoker<T> {
// If none of the invokers has a preferred signal or is picked by the loadbalancer, pick the first one available.
for (Invoker<T> invoker : invokers) {
if (invoker.isAvailable()) {
return invoker.invoke(invocation);
MockClusterInvoker<T> mockClusterInvoker = (MockClusterInvoker<T>) invoker;
if (mockClusterInvoker.isAvailable()) {
return mockClusterInvoker.invoke(invocation);
}
}
throw new RpcException("No provider available in " + invokers);

14
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockClusterInvoker.java
View File

@ -50,6 +50,10 @@ public class MockClusterInvoker<T> implements Invoker<T> {
@Override
public URL getUrl() {
return directory.getConsumerUrl();
}
public URL getRegistryUrl() {
return directory.getUrl();
}
@ -72,13 +76,13 @@ public class MockClusterInvoker<T> implements Invoker<T> {
public Result invoke(Invocation invocation) throws RpcException {
Result result = null;
String value = directory.getConsumerUrl().getMethodParameter(invocation.getMethodName(), MOCK_KEY, Boolean.FALSE.toString()).trim();
String value = getUrl().getMethodParameter(invocation.getMethodName(), MOCK_KEY, Boolean.FALSE.toString()).trim();
if (value.length() == 0 || "false".equalsIgnoreCase(value)) {
//no mock
result = this.invoker.invoke(invocation);
} else if (value.startsWith("force")) {
if (logger.isWarnEnabled()) {
logger.warn("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + directory.getConsumerUrl());
logger.warn("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + getUrl());
}
//force:direct mock
result = doMockInvoke(invocation, null);
@ -103,7 +107,7 @@ public class MockClusterInvoker<T> implements Invoker<T> {
}
if (logger.isWarnEnabled()) {
logger.warn("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + directory.getConsumerUrl(), e);
logger.warn("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + getUrl(), e);
}
result = doMockInvoke(invocation, e);
}
@ -118,7 +122,7 @@ public class MockClusterInvoker<T> implements Invoker<T> {
List<Invoker<T>> mockInvokers = selectMockInvoker(invocation);
if (CollectionUtils.isEmpty(mockInvokers)) {
minvoker = (Invoker<T>) new MockInvoker(directory.getConsumerUrl(), directory.getInterface());
minvoker = (Invoker<T>) new MockInvoker(getUrl(), directory.getInterface());
} else {
minvoker = mockInvokers.get(0);
}
@ -165,7 +169,7 @@ public class MockClusterInvoker<T> implements Invoker<T> {
} catch (RpcException e) {
if (logger.isInfoEnabled()) {
logger.info("Exception when try to invoke mock. Get mock invokers error for service:"
+ directory.getConsumerUrl().getServiceInterface() + ", method:" + invocation.getMethodName()
+ getUrl().getServiceInterface() + ", method:" + invocation.getMethodName()
+ ", will construct a new mock with 'new MockInvoker()'.", e);
}
}

5
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java
View File

@ -19,6 +19,7 @@ package org.apache.dubbo.rpc.cluster.support;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
@ -65,7 +66,6 @@ public class AbstractClusterInvokerTest {
StaticDirectory<IHelloService> dic;
RpcInvocation invocation = new RpcInvocation();
URL url = URL.valueOf("registry://localhost:9090/org.apache.dubbo.rpc.cluster.support.AbstractClusterInvokerTest.IHelloService?refer=" + URL.encode("application=abstractClusterInvokerTest"));
URL tmpUrl = url.removeParameter(REFER_KEY).removeParameter(MONITOR_KEY);
Invoker<IHelloService> invoker1;
Invoker<IHelloService> invoker2;
@ -124,7 +124,6 @@ public class AbstractClusterInvokerTest {
invokers.add(invoker1);
dic = new StaticDirectory<IHelloService>(url, invokers, null);
cluster = new AbstractClusterInvoker(dic) {
@Override
protected Result doInvoke(Invocation invocation, List invokers, LoadBalance loadbalance)
@ -226,6 +225,8 @@ public class AbstractClusterInvokerTest {
@Test
public void testCloseAvailablecheck() {
LoadBalance lb = mock(LoadBalance.class);
Map<String, String> queryMap = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
URL tmpUrl = url.addParameters(queryMap).removeParameter(MONITOR_KEY);
given(lb.select(invokers, tmpUrl, invocation)).willReturn(invoker1);
initlistsize5();