From 43c4033e11c07a59cb052a4ee7a18b95b26de707 Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Tue, 16 Aug 2022 16:36:01 +0800 Subject: [PATCH] [3.0] Fix Nacos aggregate listen (#10468) * [3.0] Fix Nacos aggregate listen * fix empty * rename --- .../nacos/NacosAggregateListener.java | 70 +++++++++++++++ .../nacos/NacosNamingServiceWrapper.java | 4 + .../dubbo/registry/nacos/NacosRegistry.java | 88 ++++++++++++++----- 3 files changed, 140 insertions(+), 22 deletions(-) create mode 100644 dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosAggregateListener.java diff --git a/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosAggregateListener.java b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosAggregateListener.java new file mode 100644 index 0000000000..5593acd915 --- /dev/null +++ b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosAggregateListener.java @@ -0,0 +1,70 @@ +/* + * 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.registry.nacos; + +import org.apache.dubbo.common.utils.ConcurrentHashSet; +import org.apache.dubbo.registry.NotifyListener; + +import com.alibaba.nacos.api.naming.pojo.Instance; + +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +public class NacosAggregateListener { + private final NotifyListener notifyListener; + private final Set serviceNames = new ConcurrentHashSet<>(); + private final Map> serviceInstances = new ConcurrentHashMap<>(); + + public NacosAggregateListener(NotifyListener notifyListener) { + this.notifyListener = notifyListener; + } + + public List saveAndAggregateAllInstances(String serviceName, List instances) { + serviceNames.add(serviceName); + serviceInstances.put(serviceName, instances); + return serviceInstances.values().stream().flatMap(List::stream).collect(Collectors.toList()); + } + + public NotifyListener getNotifyListener() { + return notifyListener; + } + + public Set getServiceNames() { + return serviceNames; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + NacosAggregateListener that = (NacosAggregateListener) o; + return Objects.equals(notifyListener, that.notifyListener) && Objects.equals(serviceNames, that.serviceNames) && Objects.equals(serviceInstances, that.serviceInstances); + } + + @Override + public int hashCode() { + return Objects.hash(notifyListener, serviceNames, serviceInstances); + } +} diff --git a/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosNamingServiceWrapper.java b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosNamingServiceWrapper.java index 6fad8f8d83..7d7a7d86fd 100644 --- a/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosNamingServiceWrapper.java +++ b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosNamingServiceWrapper.java @@ -51,6 +51,10 @@ public class NacosNamingServiceWrapper { namingService.subscribe(handleInnerSymbol(serviceName), group, eventListener); } + public void unsubscribe(String serviceName, String group, EventListener eventListener) throws NacosException { + namingService.unsubscribe(handleInnerSymbol(serviceName), group, eventListener); + } + public List getAllInstances(String serviceName, String group) throws NacosException { return namingService.getAllInstances(handleInnerSymbol(serviceName), group); } diff --git a/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java index f0ccdbe9d6..021102ced2 100644 --- a/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java +++ b/dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java @@ -39,7 +39,6 @@ import com.alibaba.nacos.api.naming.listener.EventListener; import com.alibaba.nacos.api.naming.listener.NamingEvent; import com.alibaba.nacos.api.naming.pojo.Instance; import com.alibaba.nacos.api.naming.pojo.ListView; -import com.alibaba.nacos.shaded.com.google.common.collect.Lists; import java.util.ArrayList; import java.util.Arrays; @@ -52,7 +51,6 @@ import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -134,7 +132,9 @@ public class NacosRegistry extends FailbackRegistry { */ private volatile ScheduledExecutorService scheduledExecutorService; - private final ConcurrentMap>> nacosListeners = new ConcurrentHashMap<>(); + private final Map> originToAggregateListener = new ConcurrentHashMap<>(); + + private final Map>> nacosListeners = new ConcurrentHashMap<>(); public NacosRegistry(URL url, NacosNamingServiceWrapper namingService) { super(url); @@ -203,7 +203,10 @@ public class NacosRegistry extends FailbackRegistry { @Override public void doSubscribe(final URL url, final NotifyListener listener) { - Set serviceNames = getServiceNames(url, listener); + NacosAggregateListener nacosAggregateListener = new NacosAggregateListener(listener); + originToAggregateListener.computeIfAbsent(url, k -> new ConcurrentHashMap<>()).put(listener, nacosAggregateListener); + + Set serviceNames = getServiceNames(url, nacosAggregateListener); //Set corresponding serviceNames for easy search later if (isServiceNamesWithCompatibleMode(url)) { @@ -211,14 +214,12 @@ public class NacosRegistry extends FailbackRegistry { NacosInstanceManageUtil.setCorrespondingServiceNames(serviceName, serviceNames); } } - - doSubscribe(url, listener, serviceNames); + doSubscribe(url, nacosAggregateListener, serviceNames); } - private void doSubscribe(final URL url, final NotifyListener listener, final Set serviceNames) { + private void doSubscribe(final URL url, final NacosAggregateListener listener, final Set serviceNames) { try { if (isServiceNamesWithCompatibleMode(url)) { - List allCorrespondingInstanceList = Lists.newArrayList(); /** * Get all instances with serviceNames to avoid instance overwrite and but with empty instance mentioned @@ -233,9 +234,8 @@ public class NacosRegistry extends FailbackRegistry { List instances = namingService.getAllInstances(serviceName, getUrl().getGroup(Constants.DEFAULT_GROUP)); NacosInstanceManageUtil.initOrRefreshServiceInstanceList(serviceName, instances); - allCorrespondingInstanceList.addAll(instances); + notifySubscriber(url, serviceName, listener, instances); } - notifySubscriber(url, listener, allCorrespondingInstanceList); for (String serviceName : serviceNames) { subscribeEventListener(serviceName, url, listener); } @@ -251,7 +251,7 @@ public class NacosRegistry extends FailbackRegistry { } URL subscriberURL = url.setPath(serviceInterface).addParameters(INTERFACE_KEY, serviceInterface, CHECK_KEY, String.valueOf(false)); - notifySubscriber(subscriberURL, listener, instances); + notifySubscriber(subscriberURL, serviceName, listener, instances); subscribeEventListener(serviceName, subscriberURL, listener); } } @@ -275,6 +275,26 @@ public class NacosRegistry extends FailbackRegistry { public void doUnsubscribe(URL url, NotifyListener listener) { if (isAdminProtocol(url)) { shutdownServiceNamesLookup(); + } else { + Map listenerMap = originToAggregateListener.get(url); + NacosAggregateListener nacosAggregateListener = listenerMap.remove(listener); + if (nacosAggregateListener != null) { + Set serviceNames = getServiceNames(url, nacosAggregateListener); + try { + doUnsubscribe(url, nacosAggregateListener, serviceNames); + } catch (NacosException e) { + logger.error("Failed to unsubscribe " + url + " to nacos " + getUrl() + ", cause: " + e.getMessage(), e); + } + } + if (listenerMap.isEmpty()) { + originToAggregateListener.remove(url); + } + } + } + + private void doUnsubscribe(final URL url, final NacosAggregateListener nacosAggregateListener, final Set serviceNames) throws NacosException { + for (String serviceName : serviceNames) { + unsubscribeEventListener(serviceName, url, nacosAggregateListener); } } @@ -291,7 +311,7 @@ public class NacosRegistry extends FailbackRegistry { * @param listener {@link NotifyListener} * @return non-null */ - private Set getServiceNames(URL url, NotifyListener listener) { + private Set getServiceNames(URL url, NacosAggregateListener listener) { if (isAdminProtocol(url)) { scheduleServiceNamesLookup(url, listener); return getServiceNamesForOps(url); @@ -377,7 +397,7 @@ public class NacosRegistry extends FailbackRegistry { return ADMIN_PROTOCOL.equals(url.getProtocol()); } - private void scheduleServiceNamesLookup(final URL url, final NotifyListener listener) { + private void scheduleServiceNamesLookup(final URL url, final NacosAggregateListener listener) { if (scheduledExecutorService == null) { scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(); scheduledExecutorService.scheduleAtFixedRate(() -> { @@ -529,12 +549,12 @@ public class NacosRegistry extends FailbackRegistry { return urls; } - private void subscribeEventListener(String serviceName, final URL url, final NotifyListener listener) + private void subscribeEventListener(String serviceName, final URL url, final NacosAggregateListener listener) throws NacosException { - ConcurrentMap> listeners = nacosListeners.computeIfAbsent(url, + Map> listeners = nacosListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>()); - ConcurrentMap eventListeners = listeners.computeIfAbsent(listener, + Map eventListeners = listeners.computeIfAbsent(listener, k -> new ConcurrentHashMap<>()); EventListener eventListener = eventListeners.computeIfAbsent(serviceName, @@ -545,6 +565,30 @@ public class NacosRegistry extends FailbackRegistry { eventListener); } + private void unsubscribeEventListener(String serviceName, final URL url, final NacosAggregateListener listener) throws NacosException { + Map> listenerToServiceEvent = nacosListeners.get(url); + if (listenerToServiceEvent == null) { + return; + } + Map serviceToEventMap = listenerToServiceEvent.get(listener); + if (serviceToEventMap == null) { + return; + } + EventListener eventListener = serviceToEventMap.remove(serviceName); + if (eventListener == null) { + return; + } + namingService.unsubscribe(serviceName, + getUrl().getParameter(GROUP_KEY, Constants.DEFAULT_GROUP), + eventListener); + if (serviceToEventMap.isEmpty()) { + listenerToServiceEvent.remove(listener); + } + if (listenerToServiceEvent.isEmpty()) { + nacosListeners.remove(url); + } + } + /** * Notify the Enabled {@link Instance instances} to subscriber. * @@ -552,14 +596,14 @@ public class NacosRegistry extends FailbackRegistry { * @param listener {@link NotifyListener} * @param instances all {@link Instance instances} */ - private void notifySubscriber(URL url, NotifyListener listener, Collection instances) { + private void notifySubscriber(URL url, String serviceName, NacosAggregateListener listener, Collection instances) { List enabledInstances = new LinkedList<>(instances); if (enabledInstances.size() > 0) { // Instances filterEnabledInstances(enabledInstances); } - List urls = toUrlWithEmpty(url, enabledInstances); - NacosRegistry.this.notify(url, listener, urls); + List aggregatedUrls = toUrlWithEmpty(url, listener.saveAndAggregateAllInstances(serviceName, enabledInstances)); + NacosRegistry.this.notify(url, listener.getNotifyListener(), aggregatedUrls); } /** @@ -641,9 +685,9 @@ public class NacosRegistry extends FailbackRegistry { private final URL consumerUrl; - private final NotifyListener listener; + private final NacosAggregateListener listener; - public RegistryChildListenerImpl(String serviceName, URL consumerUrl, NotifyListener listener) { + public RegistryChildListenerImpl(String serviceName, URL consumerUrl, NacosAggregateListener listener) { this.serviceName = serviceName; this.consumerUrl = consumerUrl; this.listener = listener; @@ -659,7 +703,7 @@ public class NacosRegistry extends FailbackRegistry { NacosInstanceManageUtil.initOrRefreshServiceInstanceList(serviceName, instances); instances = NacosInstanceManageUtil.getAllCorrespondingServiceInstanceList(serviceName); } - NacosRegistry.this.notifySubscriber(consumerUrl, listener, instances); + NacosRegistry.this.notifySubscriber(consumerUrl, serviceName, listener, instances); } }; }