diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/NamingSubscriberServiceV2Impl.java b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/NamingSubscriberServiceV2Impl.java index 2aec4a946..1da99ce3d 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/NamingSubscriberServiceV2Impl.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/NamingSubscriberServiceV2Impl.java @@ -25,6 +25,7 @@ import com.alibaba.nacos.naming.core.v2.client.manager.ClientManagerDelegate; import com.alibaba.nacos.naming.core.v2.event.service.ServiceEvent; import com.alibaba.nacos.naming.core.v2.index.ClientServiceIndexesManager; import com.alibaba.nacos.naming.core.v2.index.ServiceStorage; +import com.alibaba.nacos.naming.core.v2.metadata.NamingMetadataManager; import com.alibaba.nacos.naming.core.v2.pojo.Service; import com.alibaba.nacos.naming.core.v2.upgrade.UpgradeJudgement; import com.alibaba.nacos.naming.misc.SwitchDomain; @@ -60,12 +61,13 @@ public class NamingSubscriberServiceV2Impl extends SmartSubscriber implements Na public NamingSubscriberServiceV2Impl(ClientManagerDelegate clientManager, ClientServiceIndexesManager indexesManager, ServiceStorage serviceStorage, + NamingMetadataManager metadataManager, PushExecutorDelegate pushExecutor, UpgradeJudgement upgradeJudgement, SwitchDomain switchDomain) { this.clientManager = clientManager; this.indexesManager = indexesManager; this.upgradeJudgement = upgradeJudgement; this.delayTaskEngine = new PushDelayTaskExecuteEngine(clientManager, indexesManager, serviceStorage, - pushExecutor, switchDomain); + metadataManager, pushExecutor, switchDomain); NotifyCenter.registerSubscriber(this); } diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/PushDelayTaskExecuteEngine.java b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/PushDelayTaskExecuteEngine.java index b2fb8c331..f78ae89d6 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/PushDelayTaskExecuteEngine.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/PushDelayTaskExecuteEngine.java @@ -22,6 +22,7 @@ import com.alibaba.nacos.common.task.engine.NacosDelayTaskExecuteEngine; import com.alibaba.nacos.naming.core.v2.client.manager.ClientManager; import com.alibaba.nacos.naming.core.v2.index.ClientServiceIndexesManager; import com.alibaba.nacos.naming.core.v2.index.ServiceStorage; +import com.alibaba.nacos.naming.core.v2.metadata.NamingMetadataManager; import com.alibaba.nacos.naming.core.v2.pojo.Service; import com.alibaba.nacos.naming.misc.Loggers; import com.alibaba.nacos.naming.misc.NamingExecuteTaskDispatcher; @@ -41,16 +42,20 @@ public class PushDelayTaskExecuteEngine extends NacosDelayTaskExecuteEngine { private final ServiceStorage serviceStorage; + private final NamingMetadataManager metadataManager; + private final PushExecutor pushExecutor; private final SwitchDomain switchDomain; public PushDelayTaskExecuteEngine(ClientManager clientManager, ClientServiceIndexesManager indexesManager, - ServiceStorage serviceStorage, PushExecutor pushExecutor, SwitchDomain switchDomain) { + ServiceStorage serviceStorage, NamingMetadataManager metadataManager, + PushExecutor pushExecutor, SwitchDomain switchDomain) { super(PushDelayTaskExecuteEngine.class.getSimpleName(), Loggers.PUSH); this.clientManager = clientManager; this.indexesManager = indexesManager; this.serviceStorage = serviceStorage; + this.metadataManager = metadataManager; this.pushExecutor = pushExecutor; this.switchDomain = switchDomain; setDefaultTaskProcessor(new PushDelayTaskProcessor(this)); @@ -68,6 +73,10 @@ public class PushDelayTaskExecuteEngine extends NacosDelayTaskExecuteEngine { return serviceStorage; } + public NamingMetadataManager getMetadataManager() { + return metadataManager; + } + public PushExecutor getPushExecutor() { return pushExecutor; } diff --git a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/PushExecuteTask.java b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/PushExecuteTask.java index 7eb6c9847..6569cc4f2 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/PushExecuteTask.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/push/v2/task/PushExecuteTask.java @@ -20,6 +20,7 @@ import com.alibaba.nacos.api.naming.pojo.ServiceInfo; import com.alibaba.nacos.api.remote.PushCallBack; import com.alibaba.nacos.common.task.AbstractExecuteTask; import com.alibaba.nacos.naming.core.v2.client.Client; +import com.alibaba.nacos.naming.core.v2.metadata.ServiceMetadata; import com.alibaba.nacos.naming.core.v2.pojo.Service; import com.alibaba.nacos.naming.misc.Loggers; import com.alibaba.nacos.naming.pojo.Subscriber; @@ -72,7 +73,8 @@ public class PushExecuteTask extends AbstractExecuteTask { private PushDataWrapper generatePushData() { ServiceInfo serviceInfo = delayTaskEngine.getServiceStorage().getPushData(service); - serviceInfo = ServiceUtil.selectInstances(serviceInfo, false, true); + ServiceMetadata serviceMetadata = delayTaskEngine.getMetadataManager().getServiceMetadata(service).orElse(null); + serviceInfo = ServiceUtil.selectInstancesWithHealthyProtection(serviceInfo, serviceMetadata, false, true); return new PushDataWrapper(serviceInfo); } diff --git a/naming/src/main/java/com/alibaba/nacos/naming/remote/rpc/handler/SubscribeServiceRequestHandler.java b/naming/src/main/java/com/alibaba/nacos/naming/remote/rpc/handler/SubscribeServiceRequestHandler.java index ed971a576..9822199f9 100644 --- a/naming/src/main/java/com/alibaba/nacos/naming/remote/rpc/handler/SubscribeServiceRequestHandler.java +++ b/naming/src/main/java/com/alibaba/nacos/naming/remote/rpc/handler/SubscribeServiceRequestHandler.java @@ -28,6 +28,8 @@ import com.alibaba.nacos.auth.common.ActionTypes; import com.alibaba.nacos.common.utils.StringUtils; import com.alibaba.nacos.core.remote.RequestHandler; import com.alibaba.nacos.naming.core.v2.index.ServiceStorage; +import com.alibaba.nacos.naming.core.v2.metadata.NamingMetadataManager; +import com.alibaba.nacos.naming.core.v2.metadata.ServiceMetadata; import com.alibaba.nacos.naming.core.v2.pojo.Service; import com.alibaba.nacos.naming.core.v2.service.impl.EphemeralClientOperationServiceImpl; import com.alibaba.nacos.naming.pojo.Subscriber; @@ -46,11 +48,15 @@ public class SubscribeServiceRequestHandler extends RequestHandler