() {
public CanalInstance apply(String destination) {
@@ -92,6 +95,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
logger.error(String.format("stop CanalInstance[%s] has an error", entry.getKey()), e);
}
}
+ metrics.terminate();
}
public void start(final String destination) {
@@ -99,6 +103,9 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
if (!canalInstance.isStart()) {
try {
MDC.put("destination", destination);
+ if (metrics.isRunning()) {
+ metrics.register(canalInstance);
+ }
canalInstance.start();
logger.info("start CanalInstances[{}] successfully", destination);
} finally {
@@ -114,6 +121,9 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
try {
MDC.put("destination", destination);
canalInstance.stop();
+ if (metrics.isRunning()) {
+ metrics.unregister(canalInstance);
+ }
logger.info("stop CanalInstances[{}] successfully", destination);
} finally {
MDC.remove("destination");
@@ -176,7 +186,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
/**
* 获取数据
- *
+ *
*
* 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
*
@@ -188,14 +198,14 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
/**
* 获取数据，可以指定超时时间.
- *
+ *
*
* 几种case:
* a. 如果timeout为null，则采用tryGet方式，即时获取
* b. 如果timeout不为null
* 1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回
* 2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少
- *
+ *
* 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
*
*/
@@ -251,7 +261,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
/**
* 不指定 position 获取事件。canal 会记住此 client 最新的 position。
* 如果是第一次 fetch，则会从 canal 中保存的最老一条数据开始输出。
- *
+ *
*
* 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
*
@@ -264,14 +274,14 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
/**
* 不指定 position 获取事件。canal 会记住此 client 最新的 position。
* 如果是第一次 fetch，则会从 canal 中保存的最老一条数据开始输出。
- *
+ *
*
* 几种case:
* a. 如果timeout为null，则采用tryGet方式，即时获取
* b. 如果timeout不为null
* 1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回
* 2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少
- *
+ *
* 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
*
*/
@@ -342,7 +352,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
/**
* 进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
- *
+ *
*
* 注意：进行反馈时必须按照batchId的顺序进行ack(需有客户端保证)
*
@@ -492,6 +502,27 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
}
}
+ private void loadCanalMetrics() {
+ ServiceLoader providers = ServiceLoader.load(CanalMetricsProvider.class);
+ List list = new ArrayList();
+ for (CanalMetricsProvider provider : providers) {
+ list.add(provider);
+ }
+ if (!list.isEmpty()) {
+ // 发现provider, 进行初始化
+ if (list.size() > 1) {
+ logger.warn("Found more than one CanalMetricsProvider, use the first one.");
+ //报告冲突
+ for (CanalMetricsProvider p : list) {
+ logger.warn("Found CanalMetricsProvider: {}.", p.getClass().getName());
+ }
+ }
+ //默认使用第一个
+ CanalMetricsProvider provider = list.get(0);
+ this.metrics = provider.getService();
+ }
+ }
+
// ========= setter ==========
public void setCanalInstanceGenerator(CanalInstanceGenerator canalInstanceGenerator) {
diff --git a/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsProvider.java b/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsProvider.java
new file mode 100644
index 00000000..ad0ceddf
--- /dev/null
+++ b/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsProvider.java
@@ -0,0 +1,24 @@
+package com.alibaba.otter.canal.spi;
+
+/**
+ * Use java service provider mechanism to provide {@link CanalMetricsService}.
+ *
+ * Example:
+ * {@code
+ * ServiceLoader providers = ServiceLoader.load(CanalMetricsProvider.class);
+ * List list = new ArrayList();
+ * for (CanalMetricsProvider provider : providers) {
+ * list.add(provider);
+ * }
+ * }
+ *
+ * @author Chuanyi Li
+ */
+public interface CanalMetricsProvider {
+
+ /**
+ * @return Impl of {@link CanalMetricsService}
+ */
+ CanalMetricsService getService();
+
+}
diff --git a/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsService.java b/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsService.java
new file mode 100644
index 00000000..6d823be8
--- /dev/null
+++ b/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsService.java
@@ -0,0 +1,42 @@
+package com.alibaba.otter.canal.spi;
+
+import com.alibaba.otter.canal.instance.core.CanalInstance;
+
+/**
+ * Canal server/instance metrics for export.
+ *
+ * Designed to be created by service provider.
+ *
+ * @see CanalMetricsProvider
+ * @author Chuanyi Li
+ */
+public interface CanalMetricsService {
+
+ /**
+ * Initialization on canal server startup.
+ */
+ void initialize();
+
+ /**
+ * Clean-up at canal server stop phase.
+ */
+ void terminate();
+
+ /**
+ * @return {@code true} if the metrics service is running, otherwise {@code false}.
+ */
+ boolean isRunning();
+
+ /**
+ * Register instance level metrics for specified instance.
+ * @param instance {@link CanalInstance}
+ */
+ void register(CanalInstance instance);
+
+ /**
+ * Unregister instance level metrics for specified instance.
+ * @param instance {@link CanalInstance}
+ */
+ void unregister(CanalInstance instance);
+
+}
diff --git a/server/src/main/java/com/alibaba/otter/canal/spi/NopCanalMetricsService.java b/server/src/main/java/com/alibaba/otter/canal/spi/NopCanalMetricsService.java
new file mode 100644
index 00000000..0d6ad088
--- /dev/null
+++ b/server/src/main/java/com/alibaba/otter/canal/spi/NopCanalMetricsService.java
@@ -0,0 +1,38 @@
+package com.alibaba.otter.canal.spi;
+
+import com.alibaba.otter.canal.instance.core.CanalInstance;
+
+/**
+ * @author Chuanyi Li
+ */
+public class NopCanalMetricsService implements CanalMetricsService {
+
+ public static final NopCanalMetricsService NOP = new NopCanalMetricsService();
+
+ private NopCanalMetricsService() {}
+
+ @Override
+ public void initialize() {
+
+ }
+
+ @Override
+ public void terminate() {
+
+ }
+
+ @Override
+ public boolean isRunning() {
+ return false;
+ }
+
+ @Override
+ public void register(CanalInstance instance) {
+
+ }
+
+ @Override
+ public void unregister(CanalInstance instance) {
+
+ }
+}