This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new c20d9ae44 [ISSUE #3129]Refactor Registry
new ab549871a Merge pull request #3186 from mxsm/eventmesh-3129
c20d9ae44 is described below
commit c20d9ae4400e0003a44d6152a26e0c9625a1ec4f
Author: mxsm <[email protected]>
AuthorDate: Sat Feb 18 14:40:14 2023 +0800
[ISSUE #3129]Refactor Registry
---
.../eventmesh/runtime/boot/EventMeshServer.java | 28 ++++----
.../eventmesh/runtime/registry/Registry.java | 77 ++++++++++++++--------
2 files changed, 63 insertions(+), 42 deletions(-)
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
index 99046addc..7e8273545 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
@@ -19,6 +19,7 @@ package org.apache.eventmesh.runtime.boot;
import org.apache.eventmesh.common.config.CommonConfiguration;
import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.common.utils.AssertUtils;
import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
import org.apache.eventmesh.runtime.acl.Acl;
import org.apache.eventmesh.runtime.admin.controller.ClientManageController;
@@ -59,11 +60,10 @@ public class EventMeshServer {
private static final ConfigService configService =
ConfigService.getInstance();
public EventMeshServer() {
-
this.configuration =
configService.buildConfigInstance(CommonConfiguration.class);
-
+ AssertUtils.notNull(this.configuration, "configuration is null");
this.acl =
Acl.getInstance(this.configuration.getEventMeshSecurityPluginType());
- this.registry = new Registry();
+ this.registry =
Registry.getInstance(this.configuration.getEventMeshRegistryPluginType());
trace =
Trace.getInstance(this.configuration.getEventMeshTracePluginType(),
this.configuration.isEventMeshServerTraceEnable());
this.connectorResource =
ConnectorResource.getInstance(this.configuration.getEventMeshConnectorPluginType());
@@ -83,19 +83,15 @@ public class EventMeshServer {
}
public void init() throws Exception {
- if (Objects.nonNull(configuration)) {
-
- connectorResource.init();
- if (configuration.isEventMeshServerSecurityEnable()) {
- acl.init();
- }
- if (configuration.isEventMeshServerRegistryEnable()) {
- registry.init(configuration.getEventMeshRegistryPluginType());
- }
- if (configuration.isEventMeshServerTraceEnable()) {
- trace.init();
- }
-
+ connectorResource.init();
+ if (configuration.isEventMeshServerSecurityEnable()) {
+ acl.init();
+ }
+ if (configuration.isEventMeshServerRegistryEnable()) {
+ registry.init();
+ }
+ if (configuration.isEventMeshServerTraceEnable()) {
+ trace.init();
}
EventMeshTCPServer eventMeshTCPServer = null;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/registry/Registry.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/registry/Registry.java
index cb95af598..da0fd2df3 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/registry/Registry.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/registry/Registry.java
@@ -17,14 +17,17 @@
package org.apache.eventmesh.runtime.registry;
+import org.apache.eventmesh.api.exception.RegistryException;
import org.apache.eventmesh.api.registry.RegistryService;
import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo;
import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo;
import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo;
import org.apache.eventmesh.spi.EventMeshExtensionFactory;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
import lombok.extern.slf4j.Slf4j;
@@ -32,48 +35,70 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class Registry {
- private volatile boolean inited = false;
-
- private volatile boolean started = false;
+ private static final Map<String, Registry> REGISTRY_CACHE = new
HashMap<>(16);
private RegistryService registryService;
- public synchronized void init(String registryPluginType) throws Exception {
- if (!inited) {
- registryService =
EventMeshExtensionFactory.getExtension(RegistryService.class,
registryPluginType);
- if (registryService == null) {
- log.error("can't load the registryService plugin, please
check.");
- throw new RuntimeException("doesn't load the registryService
plugin, please check.");
- }
- registryService.init();
- inited = true;
+ private final AtomicBoolean inited = new AtomicBoolean(false);
+
+ private final AtomicBoolean started = new AtomicBoolean(false);
+
+ private final AtomicBoolean shutdown = new AtomicBoolean(false);
+
+ private Registry() {
+
+ }
+
+ public static Registry getInstance(String registryPluginType) {
+ return REGISTRY_CACHE.computeIfAbsent(registryPluginType, key ->
registryBuilder(key));
+ }
+
+ private static Registry registryBuilder(String registryPluginType) {
+ RegistryService registryServiceExt =
EventMeshExtensionFactory.getExtension(RegistryService.class,
registryPluginType);
+ if (registryServiceExt == null) {
+ String errorMsg = "can't load the registryService plugin, please
check.";
+ log.error(errorMsg);
+ throw new RuntimeException(errorMsg);
+ }
+ Registry registry = new Registry();
+ registry.registryService = registryServiceExt;
+
+ return registry;
+ }
+
+ public void init() throws RegistryException {
+ if (!inited.compareAndSet(false, true)) {
+ return;
}
+ registryService.init();
}
- public synchronized void start() throws Exception {
- if (!started) {
- registryService.start();
- started = true;
+ public void start() throws RegistryException {
+ if (!started.compareAndSet(false, true)) {
+ return;
}
+ registryService.start();
}
- public synchronized void shutdown() throws Exception {
- if (started) {
- registryService.shutdown();
- started = false;
- inited = false;
+ public synchronized void shutdown() throws RegistryException {
+ inited.compareAndSet(true, false);
+ started.compareAndSet(true, false);
+ if (!shutdown.compareAndSet(false, true)) {
+ return;
}
+ registryService.shutdown();
}
- public List<EventMeshDataInfo> findEventMeshInfoByCluster(String
clusterName) throws Exception {
+ public List<EventMeshDataInfo> findEventMeshInfoByCluster(String
clusterName) throws RegistryException {
return registryService.findEventMeshInfoByCluster(clusterName);
}
- public List<EventMeshDataInfo> findAllEventMeshInfo() throws Exception {
+ public List<EventMeshDataInfo> findAllEventMeshInfo() throws
RegistryException {
return registryService.findAllEventMeshInfo();
}
- public Map<String, Map<String, Integer>>
findEventMeshClientDistributionData(String clusterName, String group, String
purpose) throws Exception {
+ public Map<String, Map<String, Integer>>
findEventMeshClientDistributionData(String clusterName, String group, String
purpose)
+ throws RegistryException {
return
registryService.findEventMeshClientDistributionData(clusterName, group,
purpose);
}
@@ -81,11 +106,11 @@ public class Registry {
registryService.registerMetadata(metadata);
}
- public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo)
throws Exception {
+ public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo)
throws RegistryException {
return registryService.register(eventMeshRegisterInfo);
}
- public boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo)
throws Exception {
+ public boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo)
throws RegistryException {
return registryService.unRegister(eventMeshUnRegisterInfo);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
