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

migration optimize: unsubscribe after address received.

This commit is contained in:
ken.lj 2020-09-17 16:33:00 +08:00
parent b3cd30d43e
commit ecbf337851
3 changed files with 115 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
View File

@ -52,6 +52,7 @@ import static java.util.Collections.emptySet;
import static java.util.Collections.unmodifiableSet;
import static java.util.stream.Collectors.toSet;
import static java.util.stream.Stream.of;
import static org.apache.dubbo.common.constants.CommonConstants.CHECK_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_CHAR_SEPARATOR;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
@ -259,10 +260,13 @@ public class ServiceDiscoveryRegistry implements Registry {
public void doSubscribe(URL url, NotifyListener listener) {
writableMetadataService.subscribeURL(url);
boolean check = url.getParameter(CHECK_KEY, false);
Set<String> serviceNames = getServices(url, listener);
if (CollectionUtils.isEmpty(serviceNames)) {
throw new IllegalStateException("Should has at least one way to know which services this interface belongs to, subscription url: " + url);
if (check) {
throw new IllegalStateException("Should has at least one way to know which services this interface belongs to, subscription url: " + url);
}
}
subscribeURLs(url, listener, serviceNames);
@ -388,6 +392,42 @@ public class ServiceDiscoveryRegistry implements Registry {
.collect(toSet()));
}
// public void doSubscribe(URL url, NotifyListener listener) {
// writableMetadataService.subscribeURL(url);
//
// boolean check = url.getParameter(CHECK_KEY, false);
// Set<String> serviceNames = Collections.emptySet();
// try {
// serviceNames = getServices(url, listener);
// } catch(ServiceNameMapping.MappingException e) {
// if (!check) {
// startMappingScheduler(url, listener);
// return;
// }
// }
// if (CollectionUtils.isEmpty(serviceNames)) {
// if (check) {
// throw new IllegalStateException("Should has at least one way to know which services this interface belongs to, subscription url: " + url);
// } else {
// return;
// }
// }
//
// subscribeURLs(url, listener, serviceNames);
// }
//
// private void startMappingScheduler(URL url, NotifyListener listener) {
// scheduler.submit(() -> {
// while(true) {
// try {
// Set<String> serviceNames = serviceNameMapping.getAndListen(url);
// } catch (Exception e) {
//
// }
// }
// });
// }
/**
* Get the subscribed service names
*
@ -468,12 +508,23 @@ public class ServiceDiscoveryRegistry implements Registry {
@Override
public void onEvent(MappingChangedEvent event) {
Set<String> newApps = event.getApps();
Set<String> tempOldApps = oldApps;
oldApps = newApps;
if (CollectionUtils.isEmpty(newApps)) {
return;
}
if (!CollectionUtils.equals(oldApps, newApps) && newApps.size() >= oldApps.size()) {
doUnsubscribe(url, listener);
if (CollectionUtils.isEmpty(tempOldApps) && newApps.size() > 0) {
subscribeURLs(url, listener, newApps);
return;
}
for (String newAppName : newApps) {
if (!tempOldApps.contains(newAppName)) {
subscribeURLs(url, listener, newApps);
return;
}
}
}
}

68
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationInvoker.java
View File

@ -46,8 +46,8 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
private Class<T> type;
private RegistryProtocol registryProtocol;
private ClusterInvoker<T> invoker;
private ClusterInvoker<T> serviceDiscoveryInvoker;
private volatile ClusterInvoker<T> invoker;
private volatile ClusterInvoker<T> serviceDiscoveryInvoker;
private volatile ClusterInvoker<T> currentAvailableInvoker;
public MigrationInvoker(RegistryProtocol registryProtocol,
@ -103,13 +103,17 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
if (!forceMigrate) {
refreshServiceDiscoveryInvoker();
refreshInterfaceInvoker();
// any of the address list changes, compare these two lists.
((DynamicDirectory) invoker.getDirectory()).addInvokersChangedListener(this::compareAddresses);
((DynamicDirectory) serviceDiscoveryInvoker.getDirectory()).addInvokersChangedListener(this::compareAddresses);
this.compareAddresses();
setListener(invoker, () -> {
this.compareAddresses(invoker, serviceDiscoveryInvoker);
});
setListener(serviceDiscoveryInvoker, () -> {
this.compareAddresses(invoker, serviceDiscoveryInvoker);
});
} else {
refreshServiceDiscoveryInvoker();
destroyInterfaceInvoker();
setListener(serviceDiscoveryInvoker, () -> {
this.destroyInterfaceInvoker(this.invoker);
});
}
}
@ -143,7 +147,9 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
@Override
public synchronized void fallbackToInterfaceInvoker() {
refreshInterfaceInvoker();
destroyServiceDiscoveryInvoker();
setListener(invoker, () -> {
this.destroyServiceDiscoveryInvoker(this.serviceDiscoveryInvoker);
});
}
@Override
@ -231,7 +237,7 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
private volatile boolean invokersChanged;
private synchronized void compareAddresses() {
private synchronized void compareAddresses(ClusterInvoker<T> serviceDiscoveryInvoker, ClusterInvoker<T> invoker) {
this.invokersChanged = true;
if (logger.isDebugEnabled()) {
logger.info("" + invoker.getDirectory().getAllInvokers().size());
@ -239,25 +245,28 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
Set<MigrationAddressComparator> detectors = ExtensionLoader.getExtensionLoader(MigrationAddressComparator.class).getSupportedExtensionInstances();
if (detectors != null && detectors.stream().allMatch(migrationDetector -> migrationDetector.shouldMigrate(serviceDiscoveryInvoker, invoker))) {
discardInterfaceInvokerAddress();
discardInterfaceInvokerAddress(invoker);
} else {
discardServiceDiscoveryInvokerAddress();
discardServiceDiscoveryInvokerAddress(serviceDiscoveryInvoker);
}
}
protected synchronized void destroyServiceDiscoveryInvoker() {
this.currentAvailableInvoker = invoker;
protected synchronized void destroyServiceDiscoveryInvoker(ClusterInvoker<?> serviceDiscoveryInvoker) {
if (checkInvokerAvailable(this.invoker)) {
this.currentAvailableInvoker = this.invoker;
}
if (serviceDiscoveryInvoker != null) {
if (logger.isDebugEnabled()) {
logger.debug("Destroying instance address invokers, will not listen for address changes until re-subscribed, " + type.getName());
}
serviceDiscoveryInvoker.destroy();
serviceDiscoveryInvoker = null;
}
}
protected synchronized void discardServiceDiscoveryInvokerAddress() {
this.currentAvailableInvoker = invoker;
protected synchronized void discardServiceDiscoveryInvokerAddress(ClusterInvoker<?> serviceDiscoveryInvoker) {
if (checkInvokerAvailable(this.invoker)) {
this.currentAvailableInvoker = this.invoker;
}
if (serviceDiscoveryInvoker != null) {
if (logger.isDebugEnabled()) {
logger.debug("Discarding instance addresses, total size " + serviceDiscoveryInvoker.getDirectory().getAllInvokers().size());
@ -267,6 +276,7 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
}
protected synchronized void refreshServiceDiscoveryInvoker() {
clearListener(serviceDiscoveryInvoker);
if (needRefresh(serviceDiscoveryInvoker)) {
if (logger.isDebugEnabled()) {
logger.debug("Re-subscribing instance addresses, current interface " + type.getName());
@ -275,7 +285,20 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
}
}
private void clearListener(ClusterInvoker<T> invoker) {
if (invoker == null) return;
DynamicDirectory<T> directory = (DynamicDirectory<T>) invoker.getDirectory();
directory.setInvokersChangedListener(null);
}
private void setListener(ClusterInvoker<T> invoker, InvokersChangedListener listener) {
if (invoker == null) return;
DynamicDirectory<T> directory = (DynamicDirectory<T>) invoker.getDirectory();
directory.setInvokersChangedListener(listener);
}
protected synchronized void refreshInterfaceInvoker() {
clearListener(invoker);
if (needRefresh(invoker)) {
// FIXME invoker.destroy();
if (logger.isDebugEnabled()) {
@ -285,19 +308,22 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
}
}
protected synchronized void destroyInterfaceInvoker() {
this.currentAvailableInvoker = serviceDiscoveryInvoker;
protected synchronized void destroyInterfaceInvoker(ClusterInvoker<T> invoker) {
if (checkInvokerAvailable(this.serviceDiscoveryInvoker)) {
this.currentAvailableInvoker = this.serviceDiscoveryInvoker;
}
if (invoker != null) {
if (logger.isDebugEnabled()) {
logger.debug("Destroying interface address invokers, will not listen for address changes until re-subscribed, " + type.getName());
}
invoker.destroy();
invoker = null;
}
}
protected synchronized void discardInterfaceInvokerAddress() {
this.currentAvailableInvoker = serviceDiscoveryInvoker;
protected synchronized void discardInterfaceInvokerAddress(ClusterInvoker<T> invoker) {
if (this.serviceDiscoveryInvoker != null) {
this.currentAvailableInvoker = this.serviceDiscoveryInvoker;
}
if (invoker != null) {
if (logger.isDebugEnabled()) {
logger.debug("Discarding interface addresses, total address size " + invoker.getDirectory().getAllInvokers().size());

22
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java
View File

@ -38,9 +38,7 @@ import org.apache.dubbo.rpc.cluster.RouterFactory;
import org.apache.dubbo.rpc.cluster.directory.AbstractDirectory;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import static org.apache.dubbo.common.constants.CommonConstants.ANY_VALUE;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO;
@ -249,7 +247,7 @@ public abstract class DynamicDirectory<T> extends AbstractDirectory<T> implement
logger.warn("Failed to destroy service " + serviceKey, t);
}
invokersChangedListeners.clear();
invokersChangedListener = null;
}
@Override
@ -261,15 +259,23 @@ public abstract class DynamicDirectory<T> extends AbstractDirectory<T> implement
}
}
private Set<InvokersChangedListener> invokersChangedListeners = new HashSet<>();
private volatile InvokersChangedListener invokersChangedListener;
private volatile boolean addressChanged;
public void addInvokersChangedListener(InvokersChangedListener listener) {
invokersChangedListeners.add(listener);
public void setInvokersChangedListener(InvokersChangedListener listener) {
this.invokersChangedListener = listener;
if (addressChanged) {
invokersChangedListener.onChange();
this.addressChanged = false;
}
}
protected void invokersChanged() {
for (InvokersChangedListener l : invokersChangedListeners) {
l.onChange();
if (invokersChangedListener != null) {
invokersChangedListener.onChange();
this.addressChanged = false;
} else {
this.addressChanged = true;
}
}