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

Fix after offline and online error (#12044)

This commit is contained in:
xieshouyu 2023-04-10 17:46:41 +08:00 committed by GitHub
parent 53e009ea2f
commit d2f4ef124d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 22 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosServiceDiscovery.java
View File

@ -49,6 +49,7 @@ import static com.alibaba.nacos.api.common.Constants.DEFAULT_GROUP;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_NACOS_EXCEPTION;
import static org.apache.dubbo.common.function.ThrowableConsumer.execute;
import static org.apache.dubbo.metadata.RevisionResolver.EMPTY_REVISION;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.EXPORTED_SERVICES_REVISION_PROPERTY_NAME;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getExportedServicesRevision;
import static org.apache.dubbo.registry.nacos.util.NacosNamingServiceUtils.createNamingService;
import static org.apache.dubbo.registry.nacos.util.NacosNamingServiceUtils.getGroup;
@ -78,7 +79,7 @@ public class NacosServiceDiscovery extends AbstractServiceDiscovery {
this.namingService = createNamingService(registryURL);
// backward compatibility for 3.0.x
this.group = Boolean.parseBoolean(ConfigurationUtils.getProperty(applicationModel, NACOS_SD_USE_DEFAULT_GROUP_KEY, "false")) ?
DEFAULT_GROUP : getGroup(registryURL);
DEFAULT_GROUP : getGroup(registryURL);
}
@Override
@ -105,14 +106,15 @@ public class NacosServiceDiscovery extends AbstractServiceDiscovery {
@Override
protected void doUpdate(ServiceInstance oldServiceInstance, ServiceInstance newServiceInstance) throws RuntimeException {
if (EMPTY_REVISION.equals(getExportedServicesRevision(newServiceInstance))) {
if (EMPTY_REVISION.equals(getExportedServicesRevision(newServiceInstance))
|| EMPTY_REVISION.equals(oldServiceInstance.getMetadata().get(EXPORTED_SERVICES_REVISION_PROPERTY_NAME))) {
super.doUpdate(oldServiceInstance, newServiceInstance);
return;
}
if (!Objects.equals(oldServiceInstance.getServiceName(), newServiceInstance.getServiceName()) ||
!Objects.equals(oldServiceInstance.getAddress(), newServiceInstance.getAddress()) ||
!Objects.equals(oldServiceInstance.getPort(), newServiceInstance.getPort())) {
!Objects.equals(oldServiceInstance.getAddress(), newServiceInstance.getAddress()) ||
!Objects.equals(oldServiceInstance.getPort(), newServiceInstance.getPort())) {
// Ignore if host-ip changed. Should unregister first.
super.doUpdate(oldServiceInstance, newServiceInstance);
return;
@ -140,15 +142,15 @@ public class NacosServiceDiscovery extends AbstractServiceDiscovery {
@Override
public List<ServiceInstance> getInstances(String serviceName) throws NullPointerException {
return ThrowableFunction.execute(namingService, service ->
service.selectInstances(serviceName, group, true)
.stream().map((i) -> NacosNamingServiceUtils.toServiceInstance(registryURL, i))
.collect(Collectors.toList())
service.selectInstances(serviceName, group, true)
.stream().map((i) -> NacosNamingServiceUtils.toServiceInstance(registryURL, i))
.collect(Collectors.toList())
);
}
@Override
public void addServiceInstancesChangedListener(ServiceInstancesChangedListener listener)
throws NullPointerException, IllegalArgumentException {
throws NullPointerException, IllegalArgumentException {
// check if listener has already been added through another interface/service
if (!instanceListeners.add(listener)) {
return;
@ -225,9 +227,9 @@ public class NacosServiceDiscovery extends AbstractServiceDiscovery {
private void handleEvent(NamingEvent event, ServiceInstancesChangedListener listener) {
String serviceName = event.getServiceName();
List<ServiceInstance> serviceInstances = event.getInstances()
.stream()
.map((i) -> NacosNamingServiceUtils.toServiceInstance(registryURL, i))
.collect(Collectors.toList());
.stream()
.map((i) -> NacosNamingServiceUtils.toServiceInstance(registryURL, i))
.collect(Collectors.toList());
listener.onEvent(new ServiceInstancesChangedEvent(serviceName, serviceInstances));
}
}

15
dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscovery.java
View File

@ -28,6 +28,7 @@ import java.util.concurrent.CountDownLatch;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.x.discovery.ServiceCache;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.function.ThrowableConsumer;
import org.apache.dubbo.common.function.ThrowableFunction;
@ -44,6 +45,7 @@ import org.apache.dubbo.rpc.model.ApplicationModel;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ZOOKEEPER_EXCEPTION;
import static org.apache.dubbo.common.function.ThrowableFunction.execute;
import static org.apache.dubbo.metadata.RevisionResolver.EMPTY_REVISION;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.EXPORTED_SERVICES_REVISION_PROPERTY_NAME;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getExportedServicesRevision;
import static org.apache.dubbo.registry.zookeeper.util.CuratorFrameworkUtils.build;
import static org.apache.dubbo.registry.zookeeper.util.CuratorFrameworkUtils.buildCuratorFramework;
@ -111,7 +113,8 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery {
@Override
protected void doUpdate(ServiceInstance oldServiceInstance, ServiceInstance newServiceInstance) throws RuntimeException {
if (EMPTY_REVISION.equals(getExportedServicesRevision(newServiceInstance))) {
if (EMPTY_REVISION.equals(getExportedServicesRevision(newServiceInstance))
|| EMPTY_REVISION.equals(oldServiceInstance.getMetadata().get(EXPORTED_SERVICES_REVISION_PROPERTY_NAME))) {
super.doUpdate(oldServiceInstance, newServiceInstance);
return;
}
@ -119,7 +122,7 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery {
org.apache.curator.x.discovery.ServiceInstance<ZookeeperInstance> oldInstance = build(oldServiceInstance);
org.apache.curator.x.discovery.ServiceInstance<ZookeeperInstance> newInstance = build(newServiceInstance);
if (!Objects.equals(newInstance.getName(), oldInstance.getName()) ||
!Objects.equals(newInstance.getId(), oldInstance.getId())) {
!Objects.equals(newInstance.getId(), oldInstance.getId())) {
// Ignore if id changed. Should unregister first.
super.doUpdate(oldServiceInstance, newServiceInstance);
return;
@ -146,7 +149,7 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery {
@Override
public void addServiceInstancesChangedListener(ServiceInstancesChangedListener listener)
throws NullPointerException, IllegalArgumentException {
throws NullPointerException, IllegalArgumentException {
// check if listener has already been added through another interface/service
if (!instanceListeners.add(listener)) {
return;
@ -169,7 +172,7 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery {
watcher.getCacheInstance().close();
} catch (IOException e) {
logger.error(REGISTRY_ZOOKEEPER_EXCEPTION, "curator stop watch failed", "",
"Curator Stop service discovery watch failed. Service Name: " + serviceName);
"Curator Stop service discovery watch failed. Service Name: " + serviceName);
}
}
}
@ -190,8 +193,8 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery {
ZookeeperServiceDiscoveryChangeWatcher watcher = watcherCaches.computeIfAbsent(serviceName, name -> {
ServiceCache<ZookeeperInstance> serviceCache = serviceDiscovery.serviceCacheBuilder()
.name(name)
.build();
.name(name)
.build();
ZookeeperServiceDiscoveryChangeWatcher newer = new ZookeeperServiceDiscoveryChangeWatcher(this, serviceCache, name, latch);
serviceCache.addListener(newer);