This is an automated email from the ASF dual-hosted git repository.
albumenj pushed a commit to branch 3.2
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.2 by this push:
new 1435c23e92 Simplify event publishing (#11863)
1435c23e92 is described below
commit 1435c23e9259834f502ff9a9cb4328e341195bed
Author: wxbty <[email protected]>
AuthorDate: Tue Mar 28 16:11:14 2023 +0800
Simplify event publishing (#11863)
---
.../rpc/cluster/directory/AbstractDirectory.java | 22 +---
.../rpc/cluster/directory/StaticDirectoryTest.java | 4 +-
.../cluster/router/file/FileRouterEngineTest.java | 4 +-
.../support/AbstractClusterInvokerTest.java | 4 +-
.../support/ConnectivityValidationTest.java | 4 +-
.../support/FailoverClusterInvokerTest.java | 4 +-
.../support/wrapper/MockClusterInvokerTest.java | 4 +-
.../wrapper/MockProviderRpcExceptionTest.java | 4 +-
.../support/wrapper/ScopeClusterInvokerTest.java | 4 +-
.../common/beans/factory/ScopeBeanFactory.java | 4 +
.../org/apache/dubbo/config/ServiceConfig.java | 48 +++----
.../config/deploy/DefaultApplicationDeployer.java | 16 +--
dubbo-distribution/dubbo-all/pom.xml | 4 -
.../report/support/AbstractMetadataReport.java | 50 +++----
.../dubbo/metrics/event/ApplicationEvent.java | 36 -----
.../org/apache/dubbo/metrics/event/EmptyEvent.java | 6 +-
.../apache/dubbo/metrics/event/MethodEvent.java | 13 +-
...ventMulticaster.java => MetricsDispatcher.java} | 4 +-
.../apache/dubbo/metrics/event/MetricsEvent.java | 31 ++++-
.../dubbo/metrics/event/MetricsEventBus.java | 116 +++++++++++++++++
.../org/apache/dubbo/metrics/event/RTEvent.java | 12 +-
.../apache/dubbo/metrics/event/RequestEvent.java | 40 ------
.../event/SimpleMetricsEventMulticaster.java | 15 +--
.../{TimeCounter.java => TimeCounterEvent.java} | 17 ++-
.../event/SimpleMetricsEventMulticasterTest.java | 44 ++++---
.../metrics/MetricsScopeModelInitializer.java | 4 +-
.../collector/AggregateMetricsCollector.java | 4 +-
.../collector/HistogramMetricsCollector.java | 2 +-
.../collector/sample/MethodMetricsSampler.java | 12 +-
.../dubbo/metrics/metrics/event/RTEventTest.java | 5 +-
.../metrics/metrics/event/RequestEventTest.java | 37 ------
.../metrics/metadata/event/MetadataEvent.java | 44 +++----
.../event/StoreProviderMetadataListener.java | 2 +-
.../metadata/MetadataMetricsCollectorTest.java | 136 ++++++++++++-------
.../metrics/registry/event/RegistryEvent.java | 60 +++++----
.../collector/RegistryMetricsCollectorTest.java | 145 +++++++++++++--------
.../META-INF/native-image/reflect-config.json | 2 +-
.../registry/client/AbstractServiceDiscovery.java | 29 ++---
.../registry/client/ServiceDiscoveryRegistry.java | 24 ++--
.../listener/ServiceInstancesChangedListener.java | 39 +++---
.../registry/integration/RegistryDirectory.java | 15 +--
.../pom.xml | 4 +
42 files changed, 581 insertions(+), 493 deletions(-)
diff --git
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java
index 9109286382..21a517202f 100644
---
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java
+++
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java
@@ -27,8 +27,7 @@ import
org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.StringUtils;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
-import org.apache.dubbo.metrics.event.MetricsEvent;
+import org.apache.dubbo.metrics.event.MetricsEventBus;
import org.apache.dubbo.metrics.registry.event.RegistryEvent;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
@@ -131,8 +130,6 @@ public abstract class AbstractDirectory<T> implements
Directory<T> {
*/
private final int reconnectTaskPeriod;
- private final GlobalMetricsEventMulticaster eventMulticaster;
-
private ApplicationModel applicationModel;
public AbstractDirectory(URL url) {
@@ -187,7 +184,6 @@ public abstract class AbstractDirectory<T> implements
Directory<T> {
this.reconnectTaskPeriod = configuration.getInt(RECONNECT_TASK_PERIOD,
DEFAULT_RECONNECT_TASK_PERIOD);
setRouterChain(routerChain);
- eventMulticaster =
applicationModel.getBeanFactory().getBean(GlobalMetricsEventMulticaster.class);
}
@Override
@@ -388,15 +384,9 @@ public abstract class AbstractDirectory<T> implements
Directory<T> {
invokersToRemove.removeAll(needToRemove);
}
- private void publishMetricsEvent(MetricsEvent event) {
- if (eventMulticaster != null) {
- eventMulticaster.publishEvent(event);
- }
- }
-
@Override
public void addDisabledInvoker(Invoker<T> invoker) {
- publishMetricsEvent(new
RegistryEvent.MetricsDirectoryEvent(applicationModel,
RegistryEvent.ApplicationType.D_DISABLE));
+ MetricsEventBus.publish(new
RegistryEvent.MetricsDirectoryEvent(applicationModel,
RegistryEvent.ApplicationType.D_DISABLE));
if (invokers.contains(invoker)) {
disabledInvokers.add(invoker);
removeValidInvoker(invoker);
@@ -406,7 +396,7 @@ public abstract class AbstractDirectory<T> implements
Directory<T> {
@Override
public void recoverDisabledInvoker(Invoker<T> invoker) {
- publishMetricsEvent(new
RegistryEvent.MetricsDirectoryEvent(applicationModel,
RegistryEvent.ApplicationType.D_RECOVER_DISABLE));
+ MetricsEventBus.publish(new
RegistryEvent.MetricsDirectoryEvent(applicationModel,
RegistryEvent.ApplicationType.D_RECOVER_DISABLE));
if (disabledInvokers.remove(invoker)) {
try {
addValidInvoker(invoker);
@@ -470,7 +460,7 @@ public abstract class AbstractDirectory<T> implements
Directory<T> {
this.invokers = invokers;
refreshInvokerInternal();
this.invokersInitialized = true;
- publishMetricsEvent(new
RegistryEvent.MetricsDirectoryEvent(applicationModel,
RegistryEvent.ApplicationType.D_CURRENT, invokers.size()));
+ MetricsEventBus.publish(new
RegistryEvent.MetricsDirectoryEvent(applicationModel,
RegistryEvent.ApplicationType.D_CURRENT, invokers.size()));
}
protected void destroyInvokers() {
@@ -481,14 +471,14 @@ public abstract class AbstractDirectory<T> implements
Directory<T> {
}
private boolean addValidInvoker(Invoker<T> invoker) {
- publishMetricsEvent(new
RegistryEvent.MetricsDirectoryEvent(applicationModel,
RegistryEvent.ApplicationType.D_VALID));
+ MetricsEventBus.publish(new
RegistryEvent.MetricsDirectoryEvent(applicationModel,
RegistryEvent.ApplicationType.D_VALID));
synchronized (this.validInvokers) {
return this.validInvokers.add(invoker);
}
}
private boolean removeValidInvoker(Invoker<T> invoker) {
- publishMetricsEvent(new
RegistryEvent.MetricsDirectoryEvent(applicationModel,
RegistryEvent.ApplicationType.D_UN_VALID));
+ MetricsEventBus.publish(new
RegistryEvent.MetricsDirectoryEvent(applicationModel,
RegistryEvent.ApplicationType.D_UN_VALID));
synchronized (this.validInvokers) {
return this.validInvokers.remove(invoker);
}
diff --git
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
index 62787ccd2b..d10fd8649e 100644
---
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
+++
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
@@ -19,7 +19,7 @@ package org.apache.dubbo.rpc.cluster.directory;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.common.utils.NetUtils;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsDispatcher;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.cluster.router.MockInvoker;
@@ -62,7 +62,7 @@ class StaticDirectoryTest {
List<Invoker<String>> filteredInvokers =
router.route(invokers.clone(), URL.valueOf("consumer://" +
NetUtils.getLocalHost() + "/com.foo.BarService"), new RpcInvocation(), false,
new Holder<>());
-
ApplicationModel.defaultModel().getBeanFactory().registerBean(GlobalMetricsEventMulticaster.class);
+
ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
StaticDirectory<String> staticDirectory = new
StaticDirectory<>(filteredInvokers);
boolean isAvailable = staticDirectory.isAvailable();
Assertions.assertTrue(!isAvailable);
diff --git
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java
index 57b3162695..b025e8f8e9 100644
---
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java
+++
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java
@@ -18,7 +18,7 @@ package org.apache.dubbo.rpc.cluster.router.file;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsDispatcher;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
@@ -61,7 +61,7 @@ class FileRouterEngineTest {
@BeforeAll
public static void setUpBeforeClass() throws Exception {
-
ApplicationModel.defaultModel().getBeanFactory().registerBean(GlobalMetricsEventMulticaster.class);
+
ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
System.setProperty(ENABLE_CONNECTIVITY_VALIDATION, "false");
}
diff --git
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java
index 40bef70dfb..9ca725f8f7 100644
---
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java
+++
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java
@@ -20,7 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.StringUtils;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsDispatcher;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
@@ -107,7 +107,7 @@ class AbstractClusterInvokerTest {
@SuppressWarnings({"unchecked"})
@BeforeEach
public void setUp() throws Exception {
-
ApplicationModel.defaultModel().getBeanFactory().registerBean(GlobalMetricsEventMulticaster.class);
+
ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
Map<String, Object> attributes = new HashMap<>();
attributes.put("application", "abstractClusterInvokerTest");
url = url.putAttribute(REFER_KEY, attributes);
diff --git
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ConnectivityValidationTest.java
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ConnectivityValidationTest.java
index 59fc9cfd10..23cb42831f 100644
---
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ConnectivityValidationTest.java
+++
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ConnectivityValidationTest.java
@@ -19,7 +19,7 @@ package org.apache.dubbo.rpc.cluster.support;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.utils.CollectionUtils;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsDispatcher;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
@@ -72,7 +72,7 @@ class ConnectivityValidationTest {
@BeforeEach
public void setup() {
-
ApplicationModel.defaultModel().getBeanFactory().registerBean(GlobalMetricsEventMulticaster.class);
+
ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
invoker1 = Mockito.mock(Invoker.class);
invoker2 = Mockito.mock(Invoker.class);
invoker3 = Mockito.mock(Invoker.class);
diff --git
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java
index de20e2d262..4c99c50b35 100644
---
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java
+++
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java
@@ -17,7 +17,7 @@
package org.apache.dubbo.rpc.cluster.support;
import org.apache.dubbo.common.URL;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsDispatcher;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.AsyncRpcResult;
import org.apache.dubbo.rpc.Invocation;
@@ -67,7 +67,7 @@ class FailoverClusterInvokerTest {
@BeforeEach
public void setUp() throws Exception {
-
ApplicationModel.defaultModel().getBeanFactory().registerBean(GlobalMetricsEventMulticaster.class);
+
ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
dic = mock(Directory.class);
given(dic.getUrl()).willReturn(url);
diff --git
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockClusterInvokerTest.java
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockClusterInvokerTest.java
index 7db7129e81..b15c87324f 100644
---
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockClusterInvokerTest.java
+++
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockClusterInvokerTest.java
@@ -18,7 +18,7 @@ package org.apache.dubbo.rpc.cluster.support.wrapper;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsDispatcher;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
@@ -51,7 +51,7 @@ class MockClusterInvokerTest {
@BeforeEach
public void beforeMethod() {
-
ApplicationModel.defaultModel().getBeanFactory().registerBean(GlobalMetricsEventMulticaster.class);
+
ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
invokers.clear();
}
diff --git
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockProviderRpcExceptionTest.java
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockProviderRpcExceptionTest.java
index c0a737280f..e692aae745 100644
---
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockProviderRpcExceptionTest.java
+++
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockProviderRpcExceptionTest.java
@@ -18,7 +18,7 @@ package org.apache.dubbo.rpc.cluster.support.wrapper;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsDispatcher;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ProxyFactory;
@@ -48,7 +48,7 @@ class MockProviderRpcExceptionTest {
@BeforeEach
public void beforeMethod() {
-
ApplicationModel.defaultModel().getBeanFactory().registerBean(GlobalMetricsEventMulticaster.class);
+
ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
invokers.clear();
}
diff --git
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/ScopeClusterInvokerTest.java
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/ScopeClusterInvokerTest.java
index 889b3ab7d7..05d4401012 100644
---
a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/ScopeClusterInvokerTest.java
+++
b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/ScopeClusterInvokerTest.java
@@ -20,7 +20,7 @@ package org.apache.dubbo.rpc.cluster.support.wrapper;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsDispatcher;
import org.apache.dubbo.rpc.Exporter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
@@ -64,7 +64,7 @@ class ScopeClusterInvokerTest {
@BeforeEach
void beforeMonth() {
-
ApplicationModel.defaultModel().getBeanFactory().registerBean(GlobalMetricsEventMulticaster.class);
+
ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
}
@AfterEach
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/beans/factory/ScopeBeanFactory.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/beans/factory/ScopeBeanFactory.java
index c2b52897ca..9f8dfbe779 100644
---
a/dubbo-common/src/main/java/org/apache/dubbo/common/beans/factory/ScopeBeanFactory.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/common/beans/factory/ScopeBeanFactory.java
@@ -268,6 +268,10 @@ public class ScopeBeanFactory {
}
}
+ public boolean isDestroyed() {
+ return destroyed.get();
+ }
+
private void checkDestroyed() {
if (destroyed.get()) {
throw new IllegalStateException("ScopeBeanFactory is destroyed");
diff --git
a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java
b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java
index f628f803c8..cc553f569b 100644
---
a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java
+++
b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java
@@ -36,18 +36,15 @@ import
org.apache.dubbo.config.invoker.DelegateProviderMetaDataInvoker;
import org.apache.dubbo.config.support.Parameter;
import org.apache.dubbo.config.utils.ConfigValidationUtils;
import org.apache.dubbo.metadata.ServiceNameMapping;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
-import org.apache.dubbo.metrics.model.TimePair;
+import org.apache.dubbo.metrics.event.MetricsEventBus;
import org.apache.dubbo.metrics.registry.event.RegistryEvent;
import org.apache.dubbo.registry.client.metadata.MetadataUtils;
import org.apache.dubbo.rpc.Exporter;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
import org.apache.dubbo.rpc.ProxyFactory;
-import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.ServerService;
import org.apache.dubbo.rpc.cluster.ConfiguratorFactory;
-import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.ModuleModel;
import org.apache.dubbo.rpc.model.ModuleServiceRepository;
import org.apache.dubbo.rpc.model.ProviderModel;
@@ -427,34 +424,23 @@ public class ServiceConfig<T> extends
ServiceConfigBase<T> {
List<URL> registryURLs = ConfigValidationUtils.loadRegistries(this,
true);
- TimePair timePair = TimePair.start();
- ApplicationModel applicationModel = module.getApplicationModel();
- try {
-
applicationModel.getBeanFactory().getBean(GlobalMetricsEventMulticaster.class);
- } catch (Throwable t) {
- applicationModel = ApplicationModel.defaultModel();
- }
- GlobalMetricsEventMulticaster eventMulticaster =
applicationModel.getBeanFactory().getBean(GlobalMetricsEventMulticaster.class);
- int size = protocols.size() * registryURLs.size();
- eventMulticaster.publishEvent(new
RegistryEvent.MetricsServiceRegisterEvent(applicationModel, timePair,
getUniqueServiceName(), size));
-
- for (ProtocolConfig protocolConfig : protocols) {
- String pathKey = URL.buildKey(getContextPath(protocolConfig)
- .map(p -> p + "/" + path)
- .orElse(path), group, version);
- // stub service will use generated service name
- if (!serverService) {
- // In case user specified path, register service one more time
to map it to path.
- repository.registerService(pathKey, interfaceClass);
- }
- try {
- doExportUrlsFor1Protocol(protocolConfig, registryURLs);
- } catch (RpcException e) {
- eventMulticaster.publishErrorEvent(new
RegistryEvent.MetricsServiceRegisterEvent(applicationModel, timePair,
getUniqueServiceName(), registryURLs.size()));
- throw e;
+ MetricsEventBus.post(new
RegistryEvent.MetricsServiceRegisterEvent(module.getApplicationModel(),
getUniqueServiceName(), protocols.size() * registryURLs.size()),
+ () -> {
+ for (ProtocolConfig protocolConfig : protocols) {
+ String pathKey =
URL.buildKey(getContextPath(protocolConfig)
+ .map(p -> p + "/" + path)
+ .orElse(path), group, version);
+ // stub service will use generated service name
+ if (!serverService) {
+ // In case user specified path, register service one
more time to map it to path.
+ repository.registerService(pathKey, interfaceClass);
+ }
+ doExportUrlsFor1Protocol(protocolConfig, registryURLs);
+ }
+ return null;
}
- }
- eventMulticaster.publishFinishEvent(new
RegistryEvent.MetricsServiceRegisterEvent(applicationModel, timePair,
getUniqueServiceName(), size));
+ );
+
providerModel.setServiceUrls(urls);
}
diff --git
a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultApplicationDeployer.java
b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultApplicationDeployer.java
index 46dbf18ddc..5481b9def2 100644
---
a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultApplicationDeployer.java
+++
b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultApplicationDeployer.java
@@ -52,8 +52,7 @@ import org.apache.dubbo.metadata.report.MetadataReportFactory;
import org.apache.dubbo.metadata.report.MetadataReportInstance;
import org.apache.dubbo.metrics.collector.ConfigCenterMetricsCollector;
import org.apache.dubbo.metrics.collector.DefaultMetricsCollector;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
-import org.apache.dubbo.metrics.model.TimePair;
+import org.apache.dubbo.metrics.event.MetricsEventBus;
import org.apache.dubbo.metrics.registry.event.RegistryEvent;
import org.apache.dubbo.metrics.report.MetricsReporter;
import org.apache.dubbo.metrics.report.MetricsReporterFactory;
@@ -841,17 +840,18 @@ public class DefaultApplicationDeployer extends
AbstractDeployer<ApplicationMode
private final AtomicInteger serviceRefreshState = new AtomicInteger(0);
private void registerServiceInstance() {
- TimePair timePair = TimePair.start();
- GlobalMetricsEventMulticaster eventMulticaster =
applicationModel.getBeanFactory().getBean(GlobalMetricsEventMulticaster.class);
- eventMulticaster.publishEvent(new
RegistryEvent.MetricsApplicationRegisterEvent(applicationModel, timePair));
try {
registered = true;
-
ServiceInstanceMetadataUtils.registerMetadataAndInstance(applicationModel);
- eventMulticaster.publishFinishEvent(new
RegistryEvent.MetricsApplicationRegisterEvent(applicationModel, timePair));
+ MetricsEventBus.post(new
RegistryEvent.MetricsApplicationRegisterEvent(applicationModel),
+ () -> {
+
ServiceInstanceMetadataUtils.registerMetadataAndInstance(applicationModel);
+ return null;
+ }
+ );
} catch (Exception e) {
- eventMulticaster.publishErrorEvent(new
RegistryEvent.MetricsApplicationRegisterEvent(applicationModel, timePair));
logger.error(CONFIG_REGISTER_INSTANCE_ERROR, "configuration server
disconnected", "", "Register instance error.", e);
}
+
if (registered) {
// scheduled task for updating Metadata and ServiceInstance
asyncMetadataFuture =
frameworkExecutorRepository.getSharedScheduledExecutor().scheduleWithFixedDelay(()
-> {
diff --git a/dubbo-distribution/dubbo-all/pom.xml
b/dubbo-distribution/dubbo-all/pom.xml
index a5e7f381ea..688477c2c2 100644
--- a/dubbo-distribution/dubbo-all/pom.xml
+++ b/dubbo-distribution/dubbo-all/pom.xml
@@ -460,10 +460,6 @@
<groupId>com.alibaba.fastjson2</groupId>
<artifactId>fastjson2</artifactId>
</dependency>
- <dependency>
- <groupId>com.tdunning</groupId>
- <artifactId>t-digest</artifactId>
- </dependency>
<!-- Temporarily add this part to exclude transitive dependency -->
<dependency>
diff --git
a/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/report/support/AbstractMetadataReport.java
b/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/report/support/AbstractMetadataReport.java
index ccad3a3493..4b3d48d77b 100644
---
a/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/report/support/AbstractMetadataReport.java
+++
b/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/report/support/AbstractMetadataReport.java
@@ -30,9 +30,8 @@ import
org.apache.dubbo.metadata.report.identifier.KeyTypeEnum;
import org.apache.dubbo.metadata.report.identifier.MetadataIdentifier;
import org.apache.dubbo.metadata.report.identifier.ServiceMetadataIdentifier;
import
org.apache.dubbo.metadata.report.identifier.SubscriberMetadataIdentifier;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsEventBus;
import org.apache.dubbo.metrics.metadata.event.MetadataEvent;
-import org.apache.dubbo.metrics.model.TimePair;
import org.apache.dubbo.rpc.model.ApplicationModel;
import java.io.File;
@@ -279,29 +278,32 @@ public abstract class AbstractMetadataReport implements
MetadataReport {
}
private void storeProviderMetadataTask(MetadataIdentifier
providerMetadataIdentifier, ServiceDefinition serviceDefinition) {
- TimePair timePair = TimePair.start();
- GlobalMetricsEventMulticaster eventMulticaster =
applicationModel.getBeanFactory().getBean(GlobalMetricsEventMulticaster.class);
+
String interfaceMethodName = serviceDefinition.getCanonicalName();
- eventMulticaster.publishEvent(new
MetadataEvent.StoreProviderMetadataEvent(applicationModel, timePair,
interfaceMethodName));
- try {
- if (logger.isInfoEnabled()) {
- logger.info("store provider metadata. Identifier : " +
providerMetadataIdentifier + "; definition: " + serviceDefinition);
- }
- allMetadataReports.put(providerMetadataIdentifier,
serviceDefinition);
- failedReports.remove(providerMetadataIdentifier);
- String data = JsonUtils.getJson().toJson(serviceDefinition);
- doStoreProviderMetadata(providerMetadataIdentifier, data);
- saveProperties(providerMetadataIdentifier, data, true,
!syncReport);
- } catch (Exception e) {
- // retry again. If failed again, throw exception.
- failedReports.put(providerMetadataIdentifier, serviceDefinition);
- metadataReportRetry.startRetryTask();
- logger.error(PROXY_FAILED_EXPORT_SERVICE, "", "", "Failed to put
provider metadata " + providerMetadataIdentifier + " in " + serviceDefinition
+ ", cause: " + e.getMessage(), e);
- // fail
- eventMulticaster.publishErrorEvent(new
MetadataEvent.StoreProviderMetadataEvent(applicationModel, timePair,
interfaceMethodName));
- return;
- }
- eventMulticaster.publishFinishEvent(new
MetadataEvent.StoreProviderMetadataEvent(applicationModel, timePair,
interfaceMethodName));
+ MetadataEvent metadataEvent = new
MetadataEvent.StoreProviderMetadataEvent(applicationModel, interfaceMethodName);
+ MetricsEventBus.post(metadataEvent, () ->
+ {
+ boolean result = true;
+ try {
+ if (logger.isInfoEnabled()) {
+ logger.info("store provider metadata. Identifier : " +
providerMetadataIdentifier + "; definition: " + serviceDefinition);
+ }
+ allMetadataReports.put(providerMetadataIdentifier,
serviceDefinition);
+ failedReports.remove(providerMetadataIdentifier);
+ String data =
JsonUtils.getJson().toJson(serviceDefinition);
+ doStoreProviderMetadata(providerMetadataIdentifier, data);
+ saveProperties(providerMetadataIdentifier, data, true,
!syncReport);
+ } catch (Exception e) {
+ // retry again. If failed again, throw exception.
+ failedReports.put(providerMetadataIdentifier,
serviceDefinition);
+ metadataReportRetry.startRetryTask();
+ logger.error(PROXY_FAILED_EXPORT_SERVICE, "", "", "Failed
to put provider metadata " + providerMetadataIdentifier + " in " +
serviceDefinition + ", cause: " + e.getMessage(), e);
+ result = false;
+ }
+ return result;
+ }, aBoolean -> aBoolean
+ );
+
}
@Override
diff --git
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/ApplicationEvent.java
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/ApplicationEvent.java
deleted file mode 100644
index 07e762bf94..0000000000
---
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/ApplicationEvent.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * 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.metrics.event;
-
-public class ApplicationEvent extends MetricsEvent{
- private ApplicationEvent.Type type;
-
- public ApplicationEvent(Object source, ApplicationEvent.Type type) {
- super(source);
- this.type = type;
- }
-
- public ApplicationEvent.Type getType() {
- return type;
- }
-
- public void setType(ApplicationEvent.Type type) {
- this.type = type;
- }
-
-}
diff --git
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/EmptyEvent.java
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/EmptyEvent.java
index 3298927006..7cc8e3e1a2 100644
---
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/EmptyEvent.java
+++
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/EmptyEvent.java
@@ -17,14 +17,16 @@
package org.apache.dubbo.metrics.event;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+
/**
* EmptyEvent, do nothing.
*/
public class EmptyEvent extends MetricsEvent {
- private static final EmptyEvent empty = new EmptyEvent(new Object());
+ private static final EmptyEvent empty = new EmptyEvent(null);
- public EmptyEvent(Object source) {
+ private EmptyEvent(ApplicationModel source) {
super(source);
}
diff --git
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MethodEvent.java
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MethodEvent.java
index c876070e17..59429bf605 100644
---
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MethodEvent.java
+++
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MethodEvent.java
@@ -17,12 +17,21 @@
package org.apache.dubbo.metrics.event;
+import org.apache.dubbo.metrics.model.MethodMetric;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+
public class MethodEvent extends MetricsEvent {
private String type;
+ private final MethodMetric methodMetric;
- public MethodEvent(Object source, String type) {
- super(source);
+ public MethodEvent(ApplicationModel applicationModel, MethodMetric
methodMetric, String type) {
+ super(applicationModel);
this.type = type;
+ this.methodMetric = methodMetric;
+ }
+
+ public MethodMetric getMethodMetric() {
+ return methodMetric;
}
public String getType() {
diff --git
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/GlobalMetricsEventMulticaster.java
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MetricsDispatcher.java
similarity index 91%
rename from
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/GlobalMetricsEventMulticaster.java
rename to
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MetricsDispatcher.java
index 28f01d5c27..eebe2011f3 100644
---
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/GlobalMetricsEventMulticaster.java
+++
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MetricsDispatcher.java
@@ -27,10 +27,10 @@ import java.util.List;
/**
* Global spi event publisher
*/
-public class GlobalMetricsEventMulticaster extends
SimpleMetricsEventMulticaster {
+public class MetricsDispatcher extends SimpleMetricsEventMulticaster {
@SuppressWarnings({"rawtypes"})
- public GlobalMetricsEventMulticaster(ApplicationModel applicationModel) {
+ public MetricsDispatcher(ApplicationModel applicationModel) {
ScopeBeanFactory beanFactory = applicationModel.getBeanFactory();
ExtensionLoader<MetricsCollector> extensionLoader =
applicationModel.getExtensionLoader(MetricsCollector.class);
if (extensionLoader != null) {
diff --git
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MetricsEvent.java
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MetricsEvent.java
index 0a67ac9029..957a5a9f3b 100644
---
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MetricsEvent.java
+++
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MetricsEvent.java
@@ -18,6 +18,7 @@
package org.apache.dubbo.metrics.event;
import org.apache.dubbo.metrics.model.MethodMetric;
+import org.apache.dubbo.rpc.model.ApplicationModel;
/**
* BaseMetricsEvent.
@@ -27,17 +28,34 @@ public abstract class MetricsEvent {
/**
* Metric object. (eg. {@link MethodMetric})
*/
- protected transient Object source;
+ protected transient ApplicationModel source;
+ private boolean available = true;
- public MetricsEvent(Object source) {
+ @SuppressWarnings({"unchecked"})
+ public MetricsEvent(ApplicationModel source) {
if (source == null) {
- throw new IllegalArgumentException("null source");
+ this.source = ApplicationModel.defaultModel();
+ // Appears only in unit tests
+ this.available = false;
+ } else {
+ this.source = source;
}
+ }
+
+ public void setAvailable(boolean available) {
+ this.available = available;
+ }
+
+ public boolean isAvailable() {
+ return available;
+ }
+
+
+ public void customAfterPost(Object postResult) {
- this.source = source;
}
- public Object getSource() {
+ public ApplicationModel getSource() {
return source;
}
@@ -57,7 +75,8 @@ public abstract class MetricsEvent {
APPLICATION_INFO("APPLICATION_INFO_%s"),
NETWORK_EXCEPTION("NETWORK_EXCEPTION_%s"),
SERVICE_UNAVAILABLE("SERVICE_UNAVAILABLE_%s"),
- CODEC_EXCEPTION("CODEC_EXCEPTION_%s"),;
+ CODEC_EXCEPTION("CODEC_EXCEPTION_%s"),
+ ;
private String name;
diff --git
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MetricsEventBus.java
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MetricsEventBus.java
new file mode 100644
index 0000000000..dc83e994ab
--- /dev/null
+++
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MetricsEventBus.java
@@ -0,0 +1,116 @@
+/*
+ * 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.metrics.event;
+
+import org.apache.dubbo.common.beans.factory.ScopeBeanFactory;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+/**
+ * Dispatches events to listeners, and provides ways for listeners to register
themselves.
+ */
+public class MetricsEventBus {
+
+ /**
+ * Posts an event to all registered subscribers and only once.
+ *
+ * @param event event to post.
+ */
+ public static void publish(MetricsEvent event) {
+ if (event.getSource() == null) {
+ return;
+ }
+ ApplicationModel applicationModel = event.getSource();
+ if (applicationModel.isDestroyed()) {
+ return;
+ }
+ ScopeBeanFactory beanFactory = applicationModel.getBeanFactory();
+ if (beanFactory.isDestroyed()) {
+ return;
+ }
+ MetricsDispatcher dispatcher =
beanFactory.getBean(MetricsDispatcher.class);
+ Optional.ofNullable(dispatcher).ifPresent(d -> d.publishEvent(event));
+ }
+
+ /**
+ * Posts an event to all registered subscribers.
+ * Full lifecycle post, judging success or failure by whether there is an
exception
+ * Loop around the event target and return the original processing result
+ *
+ * @param event event to post.
+ * @param targetSupplier original processing result targetSupplier
+ */
+ public static <T> T post(MetricsEvent event, Supplier<T> targetSupplier) {
+ return post(event, targetSupplier, null);
+
+ }
+
+ /**
+ * Full lifecycle post, success and failure conditions can be customized
+ *
+ * @param event event to post.
+ * @param targetSupplier original processing result supplier
+ * @param trFunction Custom event success criteria, judged according
to the returned boolean type
+ * @param <T> Biz result type
+ * @return Biz result
+ */
+ public static <T> T post(MetricsEvent event, Supplier<T> targetSupplier,
Function<T, Boolean> trFunction) {
+ if (event.getSource() == null) {
+ return targetSupplier.get();
+ }
+ ApplicationModel applicationModel = event.getSource();
+ if (applicationModel.isDestroyed()) {
+ return targetSupplier.get();
+ }
+ ScopeBeanFactory beanFactory = applicationModel.getBeanFactory();
+ if (beanFactory.isDestroyed()) {
+ return targetSupplier.get();
+ }
+ MetricsDispatcher dispatcher =
beanFactory.getBean(MetricsDispatcher.class);
+ if (dispatcher == null) {
+ return targetSupplier.get();
+ }
+ dispatcher.publishEvent(event);
+ T result;
+ if (trFunction == null) {
+ try {
+ result = targetSupplier.get();
+ } catch (Throwable e) {
+ dispatcher.publishErrorEvent(event);
+ throw e;
+ }
+ event.customAfterPost(result);
+ dispatcher.publishFinishEvent(event);
+ } else {
+ // Custom failure status
+ result = targetSupplier.get();
+ if (trFunction.apply(result)) {
+ event.customAfterPost(result);
+ dispatcher.publishFinishEvent(event);
+ } else {
+ dispatcher.publishErrorEvent(event);
+ }
+ }
+ return result;
+ }
+
+
+}
diff --git
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/RTEvent.java
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/RTEvent.java
index eb78a4bde1..bd4f53c901 100644
---
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/RTEvent.java
+++
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/RTEvent.java
@@ -17,15 +17,19 @@
package org.apache.dubbo.metrics.event;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+
/**
* RtEvent.
*/
public class RTEvent extends MetricsEvent {
private Long rt;
+ private final Object metric;
- public RTEvent(Object source, Long rt) {
- super(source);
+ public RTEvent(ApplicationModel applicationModel, Object metric, Long rt) {
+ super(applicationModel);
this.rt = rt;
+ this.metric = metric;
}
public Long getRt() {
@@ -35,4 +39,8 @@ public class RTEvent extends MetricsEvent {
public void setRt(Long rt) {
this.rt = rt;
}
+
+ public Object getMetric() {
+ return metric;
+ }
}
diff --git
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/RequestEvent.java
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/RequestEvent.java
deleted file mode 100644
index cfa639cae2..0000000000
---
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/RequestEvent.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * 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.metrics.event;
-
-/**
- * RequestEvent.
- */
-public class RequestEvent extends MetricsEvent {
- private Type type;
-
- public RequestEvent(Object source, Type type) {
- super(source);
- this.type = type;
- }
-
- public Type getType() {
- return type;
- }
-
- public void setType(Type type) {
- this.type = type;
- }
-
-
-}
diff --git
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/SimpleMetricsEventMulticaster.java
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/SimpleMetricsEventMulticaster.java
index 64e393cc53..cc3843246f 100644
---
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/SimpleMetricsEventMulticaster.java
+++
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/SimpleMetricsEventMulticaster.java
@@ -19,7 +19,6 @@ package org.apache.dubbo.metrics.event;
import org.apache.dubbo.metrics.listener.MetricsLifeListener;
import org.apache.dubbo.metrics.listener.MetricsListener;
-import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.ArrayList;
import java.util.Collections;
@@ -53,7 +52,7 @@ public class SimpleMetricsEventMulticaster implements
MetricsEventMulticaster {
if (event instanceof EmptyEvent) {
return;
}
- if (validateIfSourceInstanceOfApplicationModel(event)) return;
+ if (validateIfApplicationConfigExist(event)) return;
for (MetricsListener listener : listeners) {
if (listener.isSupport(event)) {
listener.onEvent(event);
@@ -61,10 +60,10 @@ public class SimpleMetricsEventMulticaster implements
MetricsEventMulticaster {
}
}
- private boolean validateIfSourceInstanceOfApplicationModel(MetricsEvent
event) {
- if (event.getSource() instanceof ApplicationModel) {
+ private boolean validateIfApplicationConfigExist(MetricsEvent event) {
+ if (event.getSource() != null) {
// Check if exist application config
- return ((ApplicationModel)
event.getSource()).NotExistApplicationConfig();
+ return event.getSource().NotExistApplicationConfig();
}
return false;
}
@@ -83,12 +82,12 @@ public class SimpleMetricsEventMulticaster implements
MetricsEventMulticaster {
@SuppressWarnings({"rawtypes"})
private void publishTimeEvent(MetricsEvent event,
Consumer<MetricsLifeListener> consumer) {
- if (validateIfSourceInstanceOfApplicationModel(event)) return;
+ if (validateIfApplicationConfigExist(event)) return;
if (event instanceof EmptyEvent) {
return;
}
- if (event instanceof TimeCounter) {
- ((TimeCounter) event).getTimePair().end();
+ if (event instanceof TimeCounterEvent) {
+ ((TimeCounterEvent) event).getTimePair().end();
}
for (MetricsListener listener : listeners) {
if (listener instanceof MetricsLifeListener &&
listener.isSupport(event)) {
diff --git
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/TimeCounter.java
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/TimeCounterEvent.java
similarity index 66%
rename from
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/TimeCounter.java
rename to
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/TimeCounterEvent.java
index 780684cf1e..b4a0db1322 100644
---
a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/TimeCounter.java
+++
b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/TimeCounterEvent.java
@@ -18,11 +18,22 @@
package org.apache.dubbo.metrics.event;
import org.apache.dubbo.metrics.model.TimePair;
+import org.apache.dubbo.rpc.model.ApplicationModel;
/**
- * Mark certain types of events, allow automatic recording of start and end
times, and provide time pairs
+ * Mark certain types of events, allow automatic recording of start and end
times, and provide time pairs
*/
-public interface TimeCounter {
+public abstract class TimeCounterEvent extends MetricsEvent {
+
+ private final TimePair timePair;
+
+ public TimeCounterEvent(ApplicationModel source) {
+ super(source);
+ this.timePair = TimePair.start();
+ }
+
+ public TimePair getTimePair() {
+ return timePair;
+ }
- TimePair getTimePair();
}
diff --git
a/dubbo-metrics/dubbo-metrics-api/src/test/java/org/apache/dubbo/metrics/event/SimpleMetricsEventMulticasterTest.java
b/dubbo-metrics/dubbo-metrics-api/src/test/java/org/apache/dubbo/metrics/event/SimpleMetricsEventMulticasterTest.java
index 15f6e66f23..394034b3bb 100644
---
a/dubbo-metrics/dubbo-metrics-api/src/test/java/org/apache/dubbo/metrics/event/SimpleMetricsEventMulticasterTest.java
+++
b/dubbo-metrics/dubbo-metrics-api/src/test/java/org/apache/dubbo/metrics/event/SimpleMetricsEventMulticasterTest.java
@@ -17,30 +17,43 @@
package org.apache.dubbo.metrics.event;
+import org.apache.dubbo.config.ApplicationConfig;
+import org.apache.dubbo.config.context.ConfigManager;
import org.apache.dubbo.metrics.listener.MetricsLifeListener;
import org.apache.dubbo.metrics.listener.MetricsListener;
+import org.apache.dubbo.rpc.model.ApplicationModel;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import static
org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_MANAGEMENT_MODE_DEFAULT;
+
public class SimpleMetricsEventMulticasterTest {
private SimpleMetricsEventMulticaster eventMulticaster;
- private Object[] obj;
- MetricsEvent requestEvent;
+ private Object[] objects;
+ private final Object obj = new Object();
+ private MetricsEvent requestEvent;
@BeforeEach
public void setup() {
eventMulticaster = new SimpleMetricsEventMulticaster();
- obj = new Object[]{new Object()};
+ objects = new Object[]{obj};
eventMulticaster.addListener(new MetricsListener<MetricsEvent>() {
@Override
public void onEvent(MetricsEvent event) {
- obj[0] = new Object();
+ objects[0] = new Object();
}
});
- requestEvent = new RequestEvent(obj[0], MetricsEvent.Type.TOTAL);
-
+ ApplicationModel applicationModel = ApplicationModel.defaultModel();
+ ApplicationConfig applicationConfig = new
ApplicationConfig("provider-app");
+
applicationConfig.setExecutorManagementMode(EXECUTOR_MANAGEMENT_MODE_DEFAULT);
+
applicationModel.getApplicationConfigManager().setApplication(applicationConfig);
+ ConfigManager configManager = new ConfigManager(applicationModel);
+ configManager.setApplication(applicationConfig);
+ applicationModel.setConfigManager(configManager);
+ requestEvent = new MetricsEvent(applicationModel) {
+ };
}
@@ -48,14 +61,9 @@ public class SimpleMetricsEventMulticasterTest {
void testPublishEvent() {
// emptyEvent do nothing
- MetricsEvent emptyEvent = new EmptyEvent(obj[0]);
+ MetricsEvent emptyEvent = EmptyEvent.instance();
eventMulticaster.publishEvent(emptyEvent);
- Assertions.assertSame(emptyEvent.getSource(), obj[0]);
-
- // not empty Event change obj[]
- MetricsEvent requestEvent = new RequestEvent(obj[0],
MetricsEvent.Type.TOTAL);
- eventMulticaster.publishEvent(requestEvent);
- Assertions.assertNotSame(requestEvent.getSource(), obj[0]);
+ Assertions.assertSame(obj, objects[0]);
}
@@ -64,7 +72,7 @@ public class SimpleMetricsEventMulticasterTest {
//do nothing with no MetricsLifeListener
eventMulticaster.publishFinishEvent(requestEvent);
- Assertions.assertSame(requestEvent.getSource(), obj[0]);
+ Assertions.assertSame(obj, objects[0]);
//do onEventFinish with MetricsLifeListener
eventMulticaster.addListener((new MetricsLifeListener<MetricsEvent>() {
@@ -76,7 +84,7 @@ public class SimpleMetricsEventMulticasterTest {
@Override
public void onEventFinish(MetricsEvent event) {
- obj[0] = new Object();
+ objects[0] = new Object();
}
@Override
@@ -85,12 +93,8 @@ public class SimpleMetricsEventMulticasterTest {
}
}));
eventMulticaster.publishFinishEvent(requestEvent);
- Assertions.assertNotSame(requestEvent.getSource(), obj[0]);
+ Assertions.assertNotSame(obj, objects[0]);
}
- @Test
- void testPublishErrorEvent() {
-
- }
}
diff --git
a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/MetricsScopeModelInitializer.java
b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/MetricsScopeModelInitializer.java
index f8236f60d2..15b2806cc6 100644
---
a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/MetricsScopeModelInitializer.java
+++
b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/MetricsScopeModelInitializer.java
@@ -19,7 +19,7 @@ package org.apache.dubbo.metrics;
import org.apache.dubbo.common.beans.factory.ScopeBeanFactory;
import org.apache.dubbo.metrics.collector.DefaultMetricsCollector;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsDispatcher;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.model.ModuleModel;
@@ -36,7 +36,7 @@ public class MetricsScopeModelInitializer implements
ScopeModelInitializer {
public void initializeApplicationModel(ApplicationModel applicationModel) {
ScopeBeanFactory beanFactory = applicationModel.getBeanFactory();
beanFactory.registerBean(DefaultMetricsCollector.class);
- beanFactory.registerBean(GlobalMetricsEventMulticaster.class);
+ beanFactory.registerBean(MetricsDispatcher.class);
}
@Override
diff --git
a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java
b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java
index 664fa371de..513420a9b1 100644
---
a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java
+++
b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java
@@ -86,7 +86,7 @@ public class AggregateMetricsCollector implements
MetricsCollector, MetricsListe
}
private void onRTEvent(RTEvent event) {
- MethodMetric metric = (MethodMetric) event.getSource();
+ MethodMetric metric = (MethodMetric) event.getMetric();
Long responseTime = event.getRt();
TimeWindowQuantile quantile =
ConcurrentHashMapUtils.computeIfAbsent(rt, metric, k -> new
TimeWindowQuantile(DEFAULT_COMPRESSION, bucketNum, timeWindowSeconds));
quantile.add(responseTime);
@@ -94,7 +94,7 @@ public class AggregateMetricsCollector implements
MetricsCollector, MetricsListe
private void onRequestEvent(MethodEvent event) {
- MethodMetric metric = (MethodMetric) event.getSource();
+ MethodMetric metric = event.getMethodMetric();
String type = event.getType();
diff --git
a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/HistogramMetricsCollector.java
b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/HistogramMetricsCollector.java
index 9fc57059ac..e7137dee42 100644
---
a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/HistogramMetricsCollector.java
+++
b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/HistogramMetricsCollector.java
@@ -75,7 +75,7 @@ public class HistogramMetricsCollector implements
MetricsListener {
private void onRTEvent(RTEvent event) {
if (metricRegister != null) {
- MethodMetric metric = (MethodMetric) event.getSource();
+ MethodMetric metric = (MethodMetric) event.getMetric();
Long responseTime = event.getRt();
HistogramMetricSample sample = new
HistogramMetricSample(MetricsKey.METRIC_RT_HISTOGRAM.getNameByType(metric.getSide()),
diff --git
a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/MethodMetricsSampler.java
b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/MethodMetricsSampler.java
index a42f6c5b38..6858410995 100644
---
a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/MethodMetricsSampler.java
+++
b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/MethodMetricsSampler.java
@@ -51,7 +51,7 @@ public class MethodMetricsSampler extends
SimpleMetricsCountSampler<Invocation,
protected void countConfigure(
MetricsCountSampleConfigurer<Invocation, String, MethodMetric>
sampleConfigure) {
sampleConfigure.configureMetrics(configure -> new
MethodMetric(collector.getApplicationName(), configure.getSource()));
- sampleConfigure.configureEventHandler(configure ->
collector.getEventMulticaster().publishEvent(new
MethodEvent(configure.getMetric(),
+ sampleConfigure.configureEventHandler(configure ->
collector.getEventMulticaster().publishEvent(new
MethodEvent(collector.getApplicationModel(), configure.getMetric(),
configure.getMetricName())));
}
@@ -59,7 +59,7 @@ public class MethodMetricsSampler extends
SimpleMetricsCountSampler<Invocation,
public void rtConfigure(
MetricsCountSampleConfigurer<Invocation, String, MethodMetric>
sampleConfigure) {
sampleConfigure.configureMetrics(configure -> new
MethodMetric(collector.getApplicationName(), configure.getSource()));
- sampleConfigure.configureEventHandler(configure ->
collector.getEventMulticaster().publishEvent(new RTEvent(configure.getMetric(),
configure.getRt())));
+ sampleConfigure.configureEventHandler(configure ->
collector.getEventMulticaster().publishEvent( new
RTEvent(collector.getApplicationModel(),configure.getMetric(),
configure.getRt())));
}
@Override
@@ -99,10 +99,10 @@ public class MethodMetricsSampler extends
SimpleMetricsCountSampler<Invocation,
private <T> GaugeMetricSample<T> getGaugeMetricSample(MetricsKey
metricsKey,
- MethodMetric
methodMetric,
- MetricsCategory
metricsCategory,
- T value,
- ToDoubleFunction<T>
apply) {
+ MethodMetric
methodMetric,
+ MetricsCategory
metricsCategory,
+ T value,
+ ToDoubleFunction<T>
apply) {
return new GaugeMetricSample<>(
metricsKey.getNameByType(methodMetric.getSide()),
metricsKey.getDescription(),
diff --git
a/dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/metrics/event/RTEventTest.java
b/dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/metrics/event/RTEventTest.java
index e5134fc12e..a69c166318 100644
---
a/dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/metrics/event/RTEventTest.java
+++
b/dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/metrics/event/RTEventTest.java
@@ -19,6 +19,7 @@ package org.apache.dubbo.metrics.metrics.event;
import org.apache.dubbo.metrics.event.RTEvent;
import org.apache.dubbo.metrics.model.MethodMetric;
+import org.apache.dubbo.rpc.model.ApplicationModel;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -28,9 +29,9 @@ class RTEventTest {
void testNewEvent() {
MethodMetric metric = new MethodMetric();
Long rt = 5L;
- RTEvent event = new RTEvent(metric, rt);
+ RTEvent event = new RTEvent(ApplicationModel.defaultModel(), metric,
rt);
- Assertions.assertEquals(event.getSource(), metric);
+ Assertions.assertEquals(event.getSource(),
ApplicationModel.defaultModel());
Assertions.assertEquals(event.getRt(), rt);
}
}
diff --git
a/dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/metrics/event/RequestEventTest.java
b/dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/metrics/event/RequestEventTest.java
deleted file mode 100644
index d4031e90d5..0000000000
---
a/dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/metrics/event/RequestEventTest.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.metrics.metrics.event;
-
-import org.apache.dubbo.metrics.event.MetricsEvent;
-import org.apache.dubbo.metrics.event.RequestEvent;
-import org.apache.dubbo.metrics.model.MethodMetric;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-class RequestEventTest {
-
- @Test
- void testNewEvent() {
- MethodMetric metric = new MethodMetric();
- MetricsEvent.Type type = MetricsEvent.Type.TOTAL;
- RequestEvent event = new RequestEvent(metric, type);
-
- Assertions.assertEquals(event.getSource(), metric);
- Assertions.assertEquals(event.getType(), type);
- }
-}
diff --git
a/dubbo-metrics/dubbo-metrics-metadata/src/main/java/org/apache/dubbo/metrics/metadata/event/MetadataEvent.java
b/dubbo-metrics/dubbo-metrics-metadata/src/main/java/org/apache/dubbo/metrics/metadata/event/MetadataEvent.java
index 1b419c588d..c3ec36081a 100644
---
a/dubbo-metrics/dubbo-metrics-metadata/src/main/java/org/apache/dubbo/metrics/metadata/event/MetadataEvent.java
+++
b/dubbo-metrics/dubbo-metrics-metadata/src/main/java/org/apache/dubbo/metrics/metadata/event/MetadataEvent.java
@@ -17,45 +17,37 @@
package org.apache.dubbo.metrics.metadata.event;
-import org.apache.dubbo.metrics.event.MetricsEvent;
-import org.apache.dubbo.metrics.event.TimeCounter;
+import org.apache.dubbo.common.beans.factory.ScopeBeanFactory;
+import org.apache.dubbo.metrics.event.TimeCounterEvent;
import org.apache.dubbo.metrics.metadata.collector.MetadataMetricsCollector;
import org.apache.dubbo.metrics.model.MetricsKey;
-import org.apache.dubbo.metrics.model.TimePair;
import org.apache.dubbo.rpc.model.ApplicationModel;
/**
* Registry related events
*/
-public class MetadataEvent extends MetricsEvent implements TimeCounter {
- private final TimePair timePair;
+public class MetadataEvent extends TimeCounterEvent {
private final MetadataMetricsCollector collector;
- private final boolean available;
- public MetadataEvent(ApplicationModel applicationModel, TimePair timePair)
{
+ public MetadataEvent(ApplicationModel applicationModel) {
super(applicationModel);
- this.timePair = timePair;
- this.collector =
applicationModel.getBeanFactory().getBean(MetadataMetricsCollector.class);
- this.available = this.collector != null &&
collector.isCollectEnabled();
+ ScopeBeanFactory beanFactory = applicationModel.getBeanFactory();
+ if (beanFactory.isDestroyed()) {
+ this.collector = null;
+ } else {
+ this.collector =
beanFactory.getBean(MetadataMetricsCollector.class);
+ super.setAvailable(this.collector != null &&
collector.isCollectEnabled());
+ }
}
public ApplicationModel getSource() {
- return (ApplicationModel) source;
+ return source;
}
public MetadataMetricsCollector getCollector() {
return collector;
}
- public boolean isAvailable() {
- return available;
- }
-
- @Override
- public TimePair getTimePair() {
- return timePair;
- }
-
public enum ApplicationType {
P_TOTAL(MetricsKey.METADATA_PUSH_METRIC_NUM),
P_SUCCEED(MetricsKey.METADATA_PUSH_METRIC_NUM_SUCCEED),
@@ -119,16 +111,16 @@ public class MetadataEvent extends MetricsEvent
implements TimeCounter {
public static class PushEvent extends MetadataEvent {
- public PushEvent(ApplicationModel applicationModel, TimePair timePair)
{
- super(applicationModel, timePair);
+ public PushEvent(ApplicationModel applicationModel) {
+ super(applicationModel);
}
}
public static class SubscribeEvent extends MetadataEvent {
- public SubscribeEvent(ApplicationModel applicationModel, TimePair
timePair) {
- super(applicationModel, timePair);
+ public SubscribeEvent(ApplicationModel applicationModel) {
+ super(applicationModel);
}
}
@@ -136,8 +128,8 @@ public class MetadataEvent extends MetricsEvent implements
TimeCounter {
public static class StoreProviderMetadataEvent extends MetadataEvent {
private final String serviceKey;
- public StoreProviderMetadataEvent(ApplicationModel applicationModel,
TimePair timePair, String serviceKey) {
- super(applicationModel, timePair);
+ public StoreProviderMetadataEvent(ApplicationModel applicationModel,
String serviceKey) {
+ super(applicationModel);
this.serviceKey = serviceKey;
}
diff --git
a/dubbo-metrics/dubbo-metrics-metadata/src/main/java/org/apache/dubbo/metrics/metadata/event/StoreProviderMetadataListener.java
b/dubbo-metrics/dubbo-metrics-metadata/src/main/java/org/apache/dubbo/metrics/metadata/event/StoreProviderMetadataListener.java
index b457e1ed9e..9c69556471 100644
---
a/dubbo-metrics/dubbo-metrics-metadata/src/main/java/org/apache/dubbo/metrics/metadata/event/StoreProviderMetadataListener.java
+++
b/dubbo-metrics/dubbo-metrics-metadata/src/main/java/org/apache/dubbo/metrics/metadata/event/StoreProviderMetadataListener.java
@@ -27,7 +27,7 @@ public class StoreProviderMetadataListener implements
MetricsLifeListener<Metada
@Override
public boolean isSupport(MetricsEvent event) {
- return event instanceof MetadataEvent.StoreProviderMetadataEvent &&
((MetadataEvent) event).isAvailable();
+ return event instanceof MetadataEvent.StoreProviderMetadataEvent &&
event.isAvailable();
}
@Override
diff --git
a/dubbo-metrics/dubbo-metrics-metadata/src/test/java/org/apache/dubbo/metrics/metadata/MetadataMetricsCollectorTest.java
b/dubbo-metrics/dubbo-metrics-metadata/src/test/java/org/apache/dubbo/metrics/metadata/MetadataMetricsCollectorTest.java
index 380a99f021..447ecdb7b6 100644
---
a/dubbo-metrics/dubbo-metrics-metadata/src/test/java/org/apache/dubbo/metrics/metadata/MetadataMetricsCollectorTest.java
+++
b/dubbo-metrics/dubbo-metrics-metadata/src/test/java/org/apache/dubbo/metrics/metadata/MetadataMetricsCollectorTest.java
@@ -18,7 +18,8 @@
package org.apache.dubbo.metrics.metadata;
import org.apache.dubbo.config.ApplicationConfig;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsDispatcher;
+import org.apache.dubbo.metrics.event.MetricsEventBus;
import org.apache.dubbo.metrics.metadata.collector.MetadataMetricsCollector;
import org.apache.dubbo.metrics.metadata.event.MetadataEvent;
import org.apache.dubbo.metrics.model.MetricsKey;
@@ -36,6 +37,7 @@ import org.junit.jupiter.api.Test;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.stream.Collectors;
import static
org.apache.dubbo.common.constants.MetricsConstants.TAG_APPLICATION_NAME;
@@ -67,30 +69,42 @@ class MetadataMetricsCollectorTest {
@Test
void testPushMetrics() throws InterruptedException {
- TimePair timePair = TimePair.start();
- GlobalMetricsEventMulticaster eventMulticaster =
applicationModel.getBeanFactory().getOrRegisterBean(GlobalMetricsEventMulticaster.class);
+
applicationModel.getBeanFactory().getOrRegisterBean(MetricsDispatcher.class);
MetadataMetricsCollector collector =
applicationModel.getBeanFactory().getOrRegisterBean(MetadataMetricsCollector.class);
collector.setCollectEnabled(true);
- eventMulticaster.publishEvent(new
MetadataEvent.PushEvent(applicationModel, timePair));
- List<MetricSample> metricSamples = collector.collect();
+ MetadataEvent.PushEvent pushEvent = new
MetadataEvent.PushEvent(applicationModel);
+ MetricsEventBus.post(pushEvent,
+ () -> {
+ List<MetricSample> metricSamples = collector.collect();
- // push success +1
- Assertions.assertEquals(1, metricSamples.size());
- Assertions.assertTrue(metricSamples.get(0) instanceof
GaugeMetricSample);
- Assertions.assertEquals(metricSamples.get(0).getName(),
MetricsKey.METADATA_PUSH_METRIC_NUM.getName());
+ // push success +1
+ Assertions.assertEquals(1, metricSamples.size());
+ Assertions.assertTrue(metricSamples.get(0) instanceof
GaugeMetricSample);
+ Assertions.assertEquals(metricSamples.get(0).getName(),
MetricsKey.METADATA_PUSH_METRIC_NUM.getName());
+ return null;
+ }
+ );
- eventMulticaster.publishFinishEvent(new
MetadataEvent.PushEvent(applicationModel, timePair));
// push finish rt +1
- metricSamples = collector.collect();
+ List<MetricSample> metricSamples = collector.collect();
//num(total+success) + rt(5) = 7
Assertions.assertEquals(7, metricSamples.size());
- long c1 = timePair.calc();
- TimePair lastTimePair = TimePair.start();
- eventMulticaster.publishEvent(new
MetadataEvent.PushEvent(applicationModel, lastTimePair));
- Thread.sleep(50);
+ long c1 = pushEvent.getTimePair().calc();
+
+ pushEvent = new MetadataEvent.PushEvent(applicationModel);
+ TimePair lastTimePair = pushEvent.getTimePair();
+ MetricsEventBus.post(pushEvent,
+ () -> {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return null;
+ }, Objects::nonNull
+ );
// push error rt +1
- eventMulticaster.publishErrorEvent(new
MetadataEvent.PushEvent(applicationModel, lastTimePair));
long c2 = lastTimePair.calc();
metricSamples = collector.collect();
@@ -114,32 +128,44 @@ class MetadataMetricsCollectorTest {
}
@Test
- void testSubscribeMetrics() throws InterruptedException {
+ void testSubscribeMetrics() {
- TimePair timePair = TimePair.start();
- GlobalMetricsEventMulticaster eventMulticaster =
applicationModel.getBeanFactory().getOrRegisterBean(GlobalMetricsEventMulticaster.class);
+
applicationModel.getBeanFactory().getOrRegisterBean(MetricsDispatcher.class);
MetadataMetricsCollector collector =
applicationModel.getBeanFactory().getOrRegisterBean(MetadataMetricsCollector.class);
collector.setCollectEnabled(true);
- eventMulticaster.publishEvent(new
MetadataEvent.SubscribeEvent(applicationModel, timePair));
- List<MetricSample> metricSamples = collector.collect();
+ MetadataEvent.SubscribeEvent subscribeEvent = new
MetadataEvent.SubscribeEvent(applicationModel);
+ MetricsEventBus.post(subscribeEvent,
+ () -> {
+ List<MetricSample> metricSamples = collector.collect();
- // push success +1
- Assertions.assertEquals(1, metricSamples.size());
- Assertions.assertTrue(metricSamples.get(0) instanceof
GaugeMetricSample);
- Assertions.assertEquals(metricSamples.get(0).getName(),
MetricsKey.METADATA_SUBSCRIBE_METRIC_NUM.getName());
+ // push success +1
+ Assertions.assertEquals(1, metricSamples.size());
+ Assertions.assertTrue(metricSamples.get(0) instanceof
GaugeMetricSample);
+ Assertions.assertEquals(metricSamples.get(0).getName(),
MetricsKey.METADATA_SUBSCRIBE_METRIC_NUM.getName());
+ return null;
+ }
+ );
+ long c1 = subscribeEvent.getTimePair().calc();
- eventMulticaster.publishFinishEvent(new
MetadataEvent.SubscribeEvent(applicationModel, timePair));
// push finish rt +1
- metricSamples = collector.collect();
+ List<MetricSample> metricSamples = collector.collect();
//num(total+success) + rt(5) = 7
Assertions.assertEquals(7, metricSamples.size());
- long c1 = timePair.calc();
- TimePair lastTimePair = TimePair.start();
- eventMulticaster.publishEvent(new
MetadataEvent.SubscribeEvent(applicationModel, lastTimePair));
- Thread.sleep(50);
+ subscribeEvent = new MetadataEvent.SubscribeEvent(applicationModel);
+ TimePair lastTimePair = subscribeEvent.getTimePair();
+ MetricsEventBus.post(subscribeEvent,
+ () -> {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return null;
+ }, Objects::nonNull
+ );
+
// push error rt +1
- eventMulticaster.publishErrorEvent(new
MetadataEvent.SubscribeEvent(applicationModel, lastTimePair));
long c2 = lastTimePair.calc();
metricSamples = collector.collect();
@@ -166,33 +192,45 @@ class MetadataMetricsCollectorTest {
@Test
void testStoreProviderMetadataMetrics() throws InterruptedException {
- TimePair timePair = TimePair.start();
- GlobalMetricsEventMulticaster eventMulticaster =
applicationModel.getBeanFactory().getOrRegisterBean(GlobalMetricsEventMulticaster.class);
+
applicationModel.getBeanFactory().getOrRegisterBean(MetricsDispatcher.class);
MetadataMetricsCollector collector =
applicationModel.getBeanFactory().getOrRegisterBean(MetadataMetricsCollector.class);
collector.setCollectEnabled(true);
String serviceKey = "store.provider.test";
- eventMulticaster.publishEvent(new
MetadataEvent.StoreProviderMetadataEvent(applicationModel, timePair,
serviceKey));
- List<MetricSample> metricSamples = collector.collect();
-
- // push success +1
- Assertions.assertEquals(1, metricSamples.size());
- Assertions.assertTrue(metricSamples.get(0) instanceof
GaugeMetricSample);
- Assertions.assertEquals(metricSamples.get(0).getName(),
MetricsKey.STORE_PROVIDER_METADATA.getName());
-
Assertions.assertEquals(metricSamples.get(0).getTags().get("interface"),
serviceKey);
+ MetadataEvent metadataEvent = new
MetadataEvent.StoreProviderMetadataEvent(applicationModel, serviceKey);
+ MetricsEventBus.post(metadataEvent,
+ () -> {
+ List<MetricSample> metricSamples = collector.collect();
+
+ // push success +1
+ Assertions.assertEquals(1, metricSamples.size());
+ Assertions.assertTrue(metricSamples.get(0) instanceof
GaugeMetricSample);
+ Assertions.assertEquals(metricSamples.get(0).getName(),
MetricsKey.STORE_PROVIDER_METADATA.getName());
+
Assertions.assertEquals(metricSamples.get(0).getTags().get("interface"),
serviceKey);
+ return null;
+ }
+ );
- eventMulticaster.publishFinishEvent(new
MetadataEvent.StoreProviderMetadataEvent(applicationModel, timePair,
serviceKey));
// push finish rt +1
- metricSamples = collector.collect();
+ List<MetricSample> metricSamples = collector.collect();
//num(total+success) + rt(5) = 7
Assertions.assertEquals(7, metricSamples.size());
- long c1 = timePair.calc();
- TimePair lastTimePair = TimePair.start();
- eventMulticaster.publishEvent(new
MetadataEvent.StoreProviderMetadataEvent(applicationModel, lastTimePair,
serviceKey));
- Thread.sleep(50);
+ long c1 = metadataEvent.getTimePair().calc();
+ metadataEvent = new
MetadataEvent.StoreProviderMetadataEvent(applicationModel, serviceKey);
+ TimePair lastTimePair = metadataEvent.getTimePair();
+ MetricsEventBus.post(metadataEvent,
+ () -> {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return null;
+ }, Objects::nonNull
+ );
// push error rt +1
- eventMulticaster.publishErrorEvent(new
MetadataEvent.StoreProviderMetadataEvent(applicationModel, lastTimePair,
serviceKey));
long c2 = lastTimePair.calc();
+
metricSamples = collector.collect();
// num(total+success+error) + rt(5)
diff --git
a/dubbo-metrics/dubbo-metrics-registry/src/main/java/org/apache/dubbo/metrics/registry/event/RegistryEvent.java
b/dubbo-metrics/dubbo-metrics-registry/src/main/java/org/apache/dubbo/metrics/registry/event/RegistryEvent.java
index 5fb27ab31d..c9687df37a 100644
---
a/dubbo-metrics/dubbo-metrics-registry/src/main/java/org/apache/dubbo/metrics/registry/event/RegistryEvent.java
+++
b/dubbo-metrics/dubbo-metrics-registry/src/main/java/org/apache/dubbo/metrics/registry/event/RegistryEvent.java
@@ -17,10 +17,9 @@
package org.apache.dubbo.metrics.registry.event;
-import org.apache.dubbo.metrics.event.MetricsEvent;
-import org.apache.dubbo.metrics.event.TimeCounter;
+import org.apache.dubbo.common.beans.factory.ScopeBeanFactory;
+import org.apache.dubbo.metrics.event.TimeCounterEvent;
import org.apache.dubbo.metrics.model.MetricsKey;
-import org.apache.dubbo.metrics.model.TimePair;
import org.apache.dubbo.metrics.registry.collector.RegistryMetricsCollector;
import org.apache.dubbo.rpc.model.ApplicationModel;
@@ -29,16 +28,18 @@ import java.util.Map;
/**
* Registry related events
*/
-public class RegistryEvent extends MetricsEvent implements TimeCounter {
- private final TimePair timePair;
+public class RegistryEvent extends TimeCounterEvent {
private final RegistryMetricsCollector collector;
- private final boolean available;
- public RegistryEvent(ApplicationModel applicationModel, TimePair timePair)
{
+ public RegistryEvent(ApplicationModel applicationModel) {
super(applicationModel);
- this.timePair = timePair;
- this.collector =
applicationModel.getBeanFactory().getBean(RegistryMetricsCollector.class);
- this.available = this.collector != null &&
collector.isCollectEnabled();
+ ScopeBeanFactory beanFactory = getSource().getBeanFactory();
+ if (beanFactory.isDestroyed()) {
+ this.collector = null;
+ } else {
+ this.collector =
beanFactory.getBean(RegistryMetricsCollector.class);
+ super.setAvailable(this.collector != null &&
collector.isCollectEnabled());
+ }
}
public ApplicationModel getSource() {
@@ -49,14 +50,6 @@ public class RegistryEvent extends MetricsEvent implements
TimeCounter {
return collector;
}
- public boolean isAvailable() {
- return available;
- }
-
- @Override
- public TimePair getTimePair() {
- return timePair;
- }
public enum ApplicationType {
R_TOTAL(MetricsKey.REGISTER_METRIC_REQUESTS),
@@ -135,32 +128,37 @@ public class RegistryEvent extends MetricsEvent
implements TimeCounter {
public static class MetricsApplicationRegisterEvent extends RegistryEvent {
- public MetricsApplicationRegisterEvent(ApplicationModel
applicationModel, TimePair timePair) {
- super(applicationModel, timePair);
+ public MetricsApplicationRegisterEvent(ApplicationModel
applicationModel) {
+ super(applicationModel);
}
}
public static class MetricsSubscribeEvent extends RegistryEvent {
- public MetricsSubscribeEvent(ApplicationModel applicationModel,
TimePair timePair) {
- super(applicationModel, timePair);
+ public MetricsSubscribeEvent(ApplicationModel applicationModel) {
+ super(applicationModel);
}
}
public static class MetricsNotifyEvent extends RegistryEvent {
- private final Map<String, Integer> lastNumMap;
+ private Map<String, Integer> lastNumMap;
- public MetricsNotifyEvent(ApplicationModel applicationModel, TimePair
timePair, Map<String, Integer> lastNumMap) {
- super(applicationModel, timePair);
- this.lastNumMap = lastNumMap;
+ public MetricsNotifyEvent(ApplicationModel applicationModel) {
+ super(applicationModel);
}
public Map<String, Integer> getLastNotifyNum() {
return lastNumMap;
}
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void customAfterPost(Object postResult) {
+ this.lastNumMap = (Map<String, Integer>) postResult;
+ }
}
public static class MetricsDirectoryEvent extends RegistryEvent {
@@ -173,7 +171,7 @@ public class RegistryEvent extends MetricsEvent implements
TimeCounter {
}
public MetricsDirectoryEvent(ApplicationModel applicationModel,
ApplicationType type, int size) {
- super(applicationModel, TimePair.empty());
+ super(applicationModel);
this.type = type;
this.size = size;
}
@@ -192,8 +190,8 @@ public class RegistryEvent extends MetricsEvent implements
TimeCounter {
private final int size;
private final String serviceKey;
- public MetricsServiceRegisterEvent(ApplicationModel applicationModel,
TimePair timePair, String serviceKey, int size) {
- super(applicationModel, timePair);
+ public MetricsServiceRegisterEvent(ApplicationModel applicationModel,
String serviceKey, int size) {
+ super(applicationModel);
this.size = size;
this.serviceKey = serviceKey;
}
@@ -211,8 +209,8 @@ public class RegistryEvent extends MetricsEvent implements
TimeCounter {
private final String uniqueServiceName;
- public MetricsServiceSubscribeEvent(ApplicationModel applicationModel,
TimePair timePair, String uniqueServiceName) {
- super(applicationModel, timePair);
+ public MetricsServiceSubscribeEvent(ApplicationModel applicationModel,
String uniqueServiceName) {
+ super(applicationModel);
this.uniqueServiceName = uniqueServiceName;
}
diff --git
a/dubbo-metrics/dubbo-metrics-registry/src/test/java/org/apache/dubbo/metrics/registry/metrics/collector/RegistryMetricsCollectorTest.java
b/dubbo-metrics/dubbo-metrics-registry/src/test/java/org/apache/dubbo/metrics/registry/metrics/collector/RegistryMetricsCollectorTest.java
index 69e1b23dce..e94f1ba594 100644
---
a/dubbo-metrics/dubbo-metrics-registry/src/test/java/org/apache/dubbo/metrics/registry/metrics/collector/RegistryMetricsCollectorTest.java
+++
b/dubbo-metrics/dubbo-metrics-registry/src/test/java/org/apache/dubbo/metrics/registry/metrics/collector/RegistryMetricsCollectorTest.java
@@ -18,7 +18,8 @@
package org.apache.dubbo.metrics.registry.metrics.collector;
import org.apache.dubbo.config.ApplicationConfig;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsDispatcher;
+import org.apache.dubbo.metrics.event.MetricsEventBus;
import org.apache.dubbo.metrics.model.MetricsKey;
import org.apache.dubbo.metrics.model.MetricsKeyWrapper;
import org.apache.dubbo.metrics.model.TimePair;
@@ -35,6 +36,7 @@ import org.junit.jupiter.api.Test;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.stream.Collectors;
import static
org.apache.dubbo.common.constants.MetricsConstants.TAG_APPLICATION_NAME;
@@ -64,32 +66,45 @@ class RegistryMetricsCollectorTest {
}
@Test
- void testPushMetrics() throws InterruptedException {
+ void testRegisterMetrics() {
- TimePair timePair = TimePair.start();
- GlobalMetricsEventMulticaster eventMulticaster =
applicationModel.getBeanFactory().getOrRegisterBean(GlobalMetricsEventMulticaster.class);
+
applicationModel.getBeanFactory().getOrRegisterBean(MetricsDispatcher.class);
RegistryMetricsCollector collector =
applicationModel.getBeanFactory().getOrRegisterBean(RegistryMetricsCollector.class);
collector.setCollectEnabled(true);
- eventMulticaster.publishEvent(new
RegistryEvent.MetricsApplicationRegisterEvent(applicationModel, timePair));
- List<MetricSample> metricSamples = collector.collect();
-
- // push success +1
- Assertions.assertEquals(1, metricSamples.size());
- Assertions.assertTrue(metricSamples.get(0) instanceof
GaugeMetricSample);
+ RegistryEvent registryEvent = new
RegistryEvent.MetricsApplicationRegisterEvent(applicationModel);
+ MetricsEventBus.post(registryEvent,
+ () -> {
+ List<MetricSample> metricSamples = collector.collect();
+ // push success +1
+ Assertions.assertEquals(1, metricSamples.size());
+ Assertions.assertTrue(metricSamples.get(0) instanceof
GaugeMetricSample);
+ return null;
+ }
+ );
- eventMulticaster.publishFinishEvent(new
RegistryEvent.MetricsApplicationRegisterEvent(applicationModel, timePair));
// push finish rt +1
- metricSamples = collector.collect();
+ List<MetricSample> metricSamples = collector.collect();
//num(total+success) + rt(5) = 7
Assertions.assertEquals(7, metricSamples.size());
- long c1 = timePair.calc();
- TimePair lastTimePair = TimePair.start();
- eventMulticaster.publishEvent(new
RegistryEvent.MetricsApplicationRegisterEvent(applicationModel, lastTimePair));
- Thread.sleep(50);
+ long c1 = registryEvent.getTimePair().calc();
+
+
+ registryEvent = new
RegistryEvent.MetricsApplicationRegisterEvent(applicationModel);
+ TimePair lastTimePair = registryEvent.getTimePair();
+ MetricsEventBus.post(registryEvent,
+ () -> {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return null;
+ }, Objects::nonNull
+ );
// push error rt +1
- eventMulticaster.publishErrorEvent(new
RegistryEvent.MetricsApplicationRegisterEvent(applicationModel, lastTimePair));
long c2 = lastTimePair.calc();
+
metricSamples = collector.collect();
// num(total+success+error) + rt(5)
@@ -111,34 +126,50 @@ class RegistryMetricsCollectorTest {
Assertions.assertEquals(sampleMap.get(new
MetricsKeyWrapper(OP_TYPE_REGISTER, MetricsKey.METRIC_RT_SUM).targetKey()), c1
+ c2);
}
+
@Test
- void testServicePushMetrics() throws InterruptedException {
+ void testServicePushMetrics() {
- TimePair timePair = TimePair.start();
- GlobalMetricsEventMulticaster eventMulticaster =
applicationModel.getBeanFactory().getOrRegisterBean(GlobalMetricsEventMulticaster.class);
+
applicationModel.getBeanFactory().getOrRegisterBean(MetricsDispatcher.class);
RegistryMetricsCollector collector =
applicationModel.getBeanFactory().getOrRegisterBean(RegistryMetricsCollector.class);
collector.setCollectEnabled(true);
String serviceName = "demo.gameService";
- eventMulticaster.publishEvent(new
RegistryEvent.MetricsServiceRegisterEvent(applicationModel, timePair,
serviceName, 2));
- List<MetricSample> metricSamples = collector.collect();
+ RegistryEvent registryEvent = new
RegistryEvent.MetricsServiceRegisterEvent(applicationModel, serviceName,2);
+ MetricsEventBus.post(registryEvent,
+ () -> {
+ List<MetricSample> metricSamples = collector.collect();
- // push success +1
- Assertions.assertEquals(1, metricSamples.size());
- Assertions.assertTrue(metricSamples.get(0) instanceof
GaugeMetricSample);
+ // push success +1
+ Assertions.assertEquals(1, metricSamples.size());
+ Assertions.assertTrue(metricSamples.get(0) instanceof
GaugeMetricSample);
+ Assertions.assertEquals(metricSamples.get(0).getName(),
MetricsKey.SERVICE_REGISTER_METRIC_REQUESTS.getName());
+
Assertions.assertEquals(metricSamples.get(0).getTags().get("interface"),
serviceName);
+ return null;
+ }
+ );
- eventMulticaster.publishFinishEvent(new
RegistryEvent.MetricsServiceRegisterEvent(applicationModel, timePair,
serviceName, 2));
// push finish rt +1
- metricSamples = collector.collect();
+ List<MetricSample> metricSamples = collector.collect();
//num(total+success) + rt(5) = 7
Assertions.assertEquals(7, metricSamples.size());
- long c1 = timePair.calc();
- TimePair lastTimePair = TimePair.start();
- eventMulticaster.publishEvent(new
RegistryEvent.MetricsServiceRegisterEvent(applicationModel, lastTimePair,
serviceName, 2));
- Thread.sleep(50);
+
+ long c1 = registryEvent.getTimePair().calc();
+ registryEvent = new
RegistryEvent.MetricsServiceRegisterEvent(applicationModel, serviceName,2);
+ TimePair lastTimePair = registryEvent.getTimePair();
+ MetricsEventBus.post(registryEvent,
+ () -> {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return null;
+ }, Objects::nonNull
+ );
// push error rt +1
- eventMulticaster.publishErrorEvent(new
RegistryEvent.MetricsServiceRegisterEvent(applicationModel, lastTimePair,
serviceName, 2));
long c2 = lastTimePair.calc();
+
metricSamples = collector.collect();
// num(total+success+error) + rt(5)
@@ -162,38 +193,48 @@ class RegistryMetricsCollectorTest {
@Test
- void testServiceSubscribeMetrics() throws InterruptedException {
+ void testServiceSubscribeMetrics() {
- TimePair timePair = TimePair.start();
- GlobalMetricsEventMulticaster eventMulticaster =
applicationModel.getBeanFactory().getOrRegisterBean(GlobalMetricsEventMulticaster.class);
+
applicationModel.getBeanFactory().getOrRegisterBean(MetricsDispatcher.class);
RegistryMetricsCollector collector =
applicationModel.getBeanFactory().getOrRegisterBean(RegistryMetricsCollector.class);
collector.setCollectEnabled(true);
String serviceName = "demo.gameService";
- eventMulticaster.publishEvent(new
RegistryEvent.MetricsServiceSubscribeEvent(applicationModel, timePair,
serviceName));
-
-
- List<MetricSample> metricSamples = collector.collect();
-
- // push success +1
- Assertions.assertEquals(1, metricSamples.size());
- Assertions.assertTrue(metricSamples.get(0) instanceof
GaugeMetricSample);
- Assertions.assertEquals(metricSamples.get(0).getName(),
MetricsKey.SERVICE_SUBSCRIBE_METRIC_NUM.getName());
-
Assertions.assertEquals(metricSamples.get(0).getTags().get("interface"),
serviceName);
+ RegistryEvent subscribeEvent = new
RegistryEvent.MetricsServiceSubscribeEvent(applicationModel, serviceName);
+ MetricsEventBus.post(subscribeEvent,
+ () -> {
+ List<MetricSample> metricSamples = collector.collect();
- eventMulticaster.publishFinishEvent(new
RegistryEvent.MetricsServiceSubscribeEvent(applicationModel, timePair,
serviceName));
+ // push success +1
+ Assertions.assertEquals(1, metricSamples.size());
+ Assertions.assertTrue(metricSamples.get(0) instanceof
GaugeMetricSample);
+ Assertions.assertEquals(metricSamples.get(0).getName(),
MetricsKey.SERVICE_SUBSCRIBE_METRIC_NUM.getName());
+
Assertions.assertEquals(metricSamples.get(0).getTags().get("interface"),
serviceName);
+ return null;
+ }
+ );
// push finish rt +1
- metricSamples = collector.collect();
+ List<MetricSample> metricSamples = collector.collect();
//num(total+success) + rt(5) = 7
Assertions.assertEquals(7, metricSamples.size());
- long c1 = timePair.calc();
- TimePair lastTimePair = TimePair.start();
- eventMulticaster.publishEvent(new
RegistryEvent.MetricsServiceSubscribeEvent(applicationModel, lastTimePair,
serviceName));
- Thread.sleep(50);
+
+ long c1 = subscribeEvent.getTimePair().calc();
+ subscribeEvent = new
RegistryEvent.MetricsServiceSubscribeEvent(applicationModel, serviceName);
+ TimePair lastTimePair = subscribeEvent.getTimePair();
+ MetricsEventBus.post(subscribeEvent,
+ () -> {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return null;
+ }, Objects::nonNull
+ );
// push error rt +1
- eventMulticaster.publishErrorEvent(new
RegistryEvent.MetricsServiceSubscribeEvent(applicationModel, lastTimePair,
serviceName));
long c2 = lastTimePair.calc();
+
metricSamples = collector.collect();
// num(total+success+error) + rt(5)
diff --git
a/dubbo-native-plugin/src/main/resources/META-INF/native-image/reflect-config.json
b/dubbo-native-plugin/src/main/resources/META-INF/native-image/reflect-config.json
index 59853eab5b..36afcc0820 100644
---
a/dubbo-native-plugin/src/main/resources/META-INF/native-image/reflect-config.json
+++
b/dubbo-native-plugin/src/main/resources/META-INF/native-image/reflect-config.json
@@ -3020,7 +3020,7 @@
"allPublicMethods": true
},
{
- "name": "org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster",
+ "name": "org.apache.dubbo.metrics.event.MetricsDispatcher",
"allPublicMethods": true,
"methods": [
{
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscovery.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscovery.java
index 5e692f4538..f8b8d29a4f 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscovery.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscovery.java
@@ -26,9 +26,8 @@ import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.metadata.report.MetadataReport;
import org.apache.dubbo.metadata.report.MetadataReportInstance;
import
org.apache.dubbo.metadata.report.identifier.SubscriberMetadataIdentifier;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsEventBus;
import org.apache.dubbo.metrics.metadata.event.MetadataEvent;
-import org.apache.dubbo.metrics.model.TimePair;
import org.apache.dubbo.registry.NotifyListener;
import
org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener;
import org.apache.dubbo.registry.client.metadata.MetadataUtils;
@@ -229,18 +228,16 @@ public abstract class AbstractServiceDiscovery implements
ServiceDiscovery {
int triedTimes = 0;
while (triedTimes < 3) {
- TimePair timePair = TimePair.start();
- GlobalMetricsEventMulticaster eventMulticaster =
applicationModel.getBeanFactory().getBean(GlobalMetricsEventMulticaster.class);
- eventMulticaster.publishEvent(new
MetadataEvent.SubscribeEvent(applicationModel, timePair));
+ metadata = MetricsEventBus.post(new
MetadataEvent.PushEvent(applicationModel),
+ () -> MetadataUtils.getRemoteMetadata(revision, instances,
metadataReport),
+ result -> result != MetadataInfo.EMPTY
+ );
- metadata = MetadataUtils.getRemoteMetadata(revision,
instances, metadataReport);
if (metadata != MetadataInfo.EMPTY) {// succeeded
metadata.init();
- eventMulticaster.publishFinishEvent(new
MetadataEvent.SubscribeEvent(applicationModel, timePair));
break;
} else {// failed
- eventMulticaster.publishErrorEvent(new
MetadataEvent.SubscribeEvent(applicationModel, timePair));
if (triedTimes > 0) {
if (logger.isDebugEnabled()) {
logger.debug("Retry the " + triedTimes + " times
to get metadata for revision=" + revision);
@@ -361,19 +358,15 @@ public abstract class AbstractServiceDiscovery implements
ServiceDiscovery {
if (metadataInfo == null) {
return;
}
- TimePair timePair = TimePair.start();
- GlobalMetricsEventMulticaster eventMulticaster =
applicationModel.getBeanFactory().getBean(GlobalMetricsEventMulticaster.class);
if (metadataReport != null) {
SubscriberMetadataIdentifier identifier = new
SubscriberMetadataIdentifier(serviceName, metadataInfo.getRevision());
if ((DEFAULT_METADATA_STORAGE_TYPE.equals(metadataType) &&
metadataReport.shouldReportMetadata()) ||
REMOTE_METADATA_STORAGE_TYPE.equals(metadataType)) {
- try {
- eventMulticaster.publishEvent(new
MetadataEvent.PushEvent(applicationModel, timePair));
- metadataReport.publishAppMetadata(identifier,
metadataInfo);
- eventMulticaster.publishFinishEvent(new
MetadataEvent.PushEvent(applicationModel, timePair));
- } catch (IllegalStateException e) {
- eventMulticaster.publishErrorEvent(new
MetadataEvent.PushEvent(applicationModel, timePair));
- throw e;
- }
+ MetricsEventBus.post(new
MetadataEvent.PushEvent(applicationModel),
+ () ->
+ {
+ metadataReport.publishAppMetadata(identifier,
metadataInfo);
+ return null;
+ });
}
}
MetadataInfo clonedMetadataInfo = metadataInfo.clone();
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
index 0714987b38..e689b2ecc8 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
@@ -25,8 +25,7 @@ import org.apache.dubbo.metadata.AbstractServiceNameMapping;
import org.apache.dubbo.metadata.MappingChangedEvent;
import org.apache.dubbo.metadata.MappingListener;
import org.apache.dubbo.metadata.ServiceNameMapping;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
-import org.apache.dubbo.metrics.model.TimePair;
+import org.apache.dubbo.metrics.event.MetricsEventBus;
import org.apache.dubbo.metrics.registry.event.RegistryEvent;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent;
@@ -36,7 +35,6 @@ import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -326,25 +324,19 @@ public class ServiceDiscoveryRegistry extends
FailbackRegistry {
serviceListeners.put(serviceNamesKey,
serviceInstancesChangedListener);
}
- ApplicationModel applicationModel =
Optional.ofNullable(url.getApplicationModel()).orElse(ApplicationModel.defaultModel())
;
- GlobalMetricsEventMulticaster eventMulticaster =
applicationModel.getBeanFactory().getBean(GlobalMetricsEventMulticaster.class);
- TimePair timePair = TimePair.start();
-
- eventMulticaster.publishEvent(new
RegistryEvent.MetricsServiceSubscribeEvent(applicationModel, timePair,
serviceKey));
if (!serviceInstancesChangedListener.isDestroyed()) {
listener.addServiceListener(serviceInstancesChangedListener);
serviceInstancesChangedListener.addListenerAndNotify(url,
listener);
- try {
-
serviceDiscovery.addServiceInstancesChangedListener(serviceInstancesChangedListener);
- } catch (Throwable t) {
- eventMulticaster.publishErrorEvent(new
RegistryEvent.MetricsServiceSubscribeEvent(applicationModel, timePair,
serviceKey));
- throw t;
- }
- eventMulticaster.publishFinishEvent(new
RegistryEvent.MetricsServiceSubscribeEvent(applicationModel, timePair,
serviceKey));
+ ServiceInstancesChangedListener
finalServiceInstancesChangedListener = serviceInstancesChangedListener;
+ MetricsEventBus.post(new
RegistryEvent.MetricsServiceSubscribeEvent(url.getApplicationModel(),
serviceKey),
+ () -> {
+
serviceDiscovery.addServiceInstancesChangedListener(finalServiceInstancesChangedListener);
+ return null;
+ }
+ );
} else {
logger.info(String.format("Listener of %s has been destroyed
by another thread.", serviceNamesKey));
serviceListeners.remove(serviceNamesKey);
- eventMulticaster.publishErrorEvent(new
RegistryEvent.MetricsSubscribeEvent(applicationModel, timePair));
}
} finally {
appSubscriptionLock.unlock();
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
index 0859c1e9e2..655e72fc19 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
@@ -19,7 +19,6 @@ package org.apache.dubbo.registry.client.event.listener;
import org.apache.dubbo.common.ProtocolServiceKey;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.URLBuilder;
-import org.apache.dubbo.common.beans.factory.ScopeBeanFactory;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
@@ -28,8 +27,7 @@ import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.metadata.MetadataInfo.ServiceInfo;
-import org.apache.dubbo.metrics.event.SimpleMetricsEventMulticaster;
-import org.apache.dubbo.metrics.model.TimePair;
+import org.apache.dubbo.metrics.event.MetricsEventBus;
import org.apache.dubbo.metrics.registry.event.RegistryEvent;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.client.DefaultServiceInstance;
@@ -405,25 +403,24 @@ public class ServiceInstancesChangedListener {
*/
protected void notifyAddressChanged() {
- ScopeBeanFactory beanFactory =
applicationModel.getFrameworkModel().getBeanFactory();
- SimpleMetricsEventMulticaster eventMulticaster =
beanFactory.getOrRegisterBean(SimpleMetricsEventMulticaster.class);
-
- TimePair timePair = TimePair.start();
- eventMulticaster.publishEvent(new
RegistryEvent.MetricsNotifyEvent(applicationModel, timePair, null));
- Map<String, Integer> lastNumMap = new HashMap<>();
- // 1 different services
- listeners.forEach((serviceKey, listenerSet) -> {
- // 2 multiple subscription listener of the same service
- for (NotifyListenerWithKey listenerWithKey : listenerSet) {
- NotifyListener notifyListener =
listenerWithKey.getNotifyListener();
-
- List<URL> urls =
toUrlsWithEmpty(getAddresses(listenerWithKey.getProtocolServiceKey(),
notifyListener.getConsumerUrl()));
- logger.info("Notify service " +
listenerWithKey.getProtocolServiceKey() + " with urls " + urls.size());
- notifyListener.notify(urls);
- lastNumMap.put(serviceKey, urls.size());
+ MetricsEventBus.post(new
RegistryEvent.MetricsNotifyEvent(applicationModel),
+ () -> {
+ Map<String, Integer> lastNumMap = new HashMap<>();
+ // 1 different services
+ listeners.forEach((serviceKey, listenerSet) -> {
+ // 2 multiple subscription listener of the same service
+ for (NotifyListenerWithKey listenerWithKey : listenerSet) {
+ NotifyListener notifyListener =
listenerWithKey.getNotifyListener();
+
+ List<URL> urls =
toUrlsWithEmpty(getAddresses(listenerWithKey.getProtocolServiceKey(),
notifyListener.getConsumerUrl()));
+ logger.info("Notify service " +
listenerWithKey.getProtocolServiceKey() + " with urls " + urls.size());
+ notifyListener.notify(urls);
+ lastNumMap.put(serviceKey, urls.size());
+ }
+ });
+ return lastNumMap;
}
- });
- eventMulticaster.publishFinishEvent(new
RegistryEvent.MetricsNotifyEvent(applicationModel, timePair, lastNumMap));
+ );
}
diff --git
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java
index fe85a484ab..53e6bd3706 100644
---
a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java
+++
b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java
@@ -30,8 +30,7 @@ import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.common.utils.UrlUtils;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
-import org.apache.dubbo.metrics.model.TimePair;
+import org.apache.dubbo.metrics.event.MetricsEventBus;
import org.apache.dubbo.metrics.registry.event.RegistryEvent;
import org.apache.dubbo.registry.AddressListener;
import org.apache.dubbo.remoting.Constants;
@@ -120,12 +119,12 @@ public class RegistryDirectory<T> extends
DynamicDirectory<T> {
@Override
public void subscribe(URL url) {
ApplicationModel applicationModel = url.getApplicationModel();
- GlobalMetricsEventMulticaster eventMulticaster =
applicationModel.getBeanFactory().getBean(GlobalMetricsEventMulticaster.class);
- TimePair timePair = TimePair.start();
-
- eventMulticaster.publishEvent(new
RegistryEvent.MetricsSubscribeEvent(applicationModel, timePair));
- super.subscribe(url);
- eventMulticaster.publishFinishEvent(new
RegistryEvent.MetricsSubscribeEvent(applicationModel, timePair));
+ MetricsEventBus.post(new
RegistryEvent.MetricsSubscribeEvent(applicationModel),() ->
+ {
+ super.subscribe(url);
+ return null;
+ }
+ );
if
(moduleModel.getModelEnvironment().getConfiguration().convert(Boolean.class,
org.apache.dubbo.registry.Constants.ENABLE_CONFIGURATION_LISTEN, true)) {
consumerConfigurationListener.addNotifyListener(this);
referenceConfigurationListener = new
ReferenceConfigurationListener(moduleModel, this, url);
diff --git a/dubbo-spring-boot/dubbo-spring-boot-observability-starter/pom.xml
b/dubbo-spring-boot/dubbo-spring-boot-observability-starter/pom.xml
index 346953efe7..b5c00de729 100644
--- a/dubbo-spring-boot/dubbo-spring-boot-observability-starter/pom.xml
+++ b/dubbo-spring-boot/dubbo-spring-boot-observability-starter/pom.xml
@@ -69,6 +69,10 @@
<groupId>io.micrometer</groupId>
<artifactId>micrometer-registry-prometheus</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.tdunning</groupId>
+ <artifactId>t-digest</artifactId>
+ </dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure</artifactId>
