Merge branch 'master' into master

2026-11-04 00:30:15 +08:00 · 2019-01-17 10:48:14 +08:00
parent d2bbe9b061 171bedf488
commit 4beb1e6855
4 changed files with 458 additions and 432 deletions
--- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
+++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
@@ -88,6 +88,7 @@ public class RdbSyncService {
     * @param function 回调方法
     */
    public void sync(List<Dml> dmls, Function<Dml, Boolean> function) {
+        try {
            boolean toExecute = false;
            for (Dml dml : dmls) {
                if (!toExecute) {
@@ -122,13 +123,14 @@ public class RdbSyncService {
                    }
                });
            }
-
+        } finally {
            for (BatchExecutor batchExecutor : batchExecutors) {
                if (batchExecutor != null) {
                    batchExecutor.close();
                }
            }
        }
+    }

    /**
     * 批量同步
--- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
+++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
@@ -68,6 +68,7 @@ public class CanalController {

    private CanalInstanceGenerator                   instanceGenerator;
    private ZkClientx                                zkclientx;
+    private String                                   zkServers;

    private CanalMQStarter                           canalMQStarter;

@@ -132,6 +133,7 @@ public class CanalController {
        }
        final String zkServers = getProperty(properties, CanalConstants.CANAL_ZKSERVERS);
        if (StringUtils.isNotEmpty(zkServers)) {
+            this.zkServers = zkServers;
            zkclientx = ZkClientx.getZkClient(zkServers);
            // 初始化系统目录
            zkclientx.createPersistent(ZookeeperPathUtils.DESTINATION_ROOT_NODE, true);
@@ -140,7 +142,8 @@ public class CanalController {

        final ServerRunningData serverData = new ServerRunningData(cid, ip + ":" + port);
        ServerRunningMonitors.setServerData(serverData);
-        ServerRunningMonitors.setRunningMonitors(MigrateMap.makeComputingMap(new Function<String, ServerRunningMonitor>() {
+        ServerRunningMonitors
+            .setRunningMonitors(MigrateMap.makeComputingMap(new Function<String, ServerRunningMonitor>() {

                public ServerRunningMonitor apply(final String destination) {
                    ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData);
@@ -168,8 +171,8 @@ public class CanalController {
                        public void processStart() {
                            try {
                                if (zkclientx != null) {
-                                final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":"
-                                                                                                              + port);
+                                    final String path = ZookeeperPathUtils.getDestinationClusterNode(destination,
+                                        ip + ":" + port);
                                    initCid(path);
                                    zkclientx.subscribeStateChanges(new IZkStateListener() {

@@ -196,8 +199,8 @@ public class CanalController {
                            try {
                                MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
                                if (zkclientx != null) {
-                                final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":"
-                                                                                                              + port);
+                                    final String path = ZookeeperPathUtils.getDestinationClusterNode(destination,
+                                        ip + ":" + port);
                                    releaseCid(path);
                                }
                            } finally {
@@ -265,7 +268,8 @@ public class CanalController {
            instanceConfigMonitors = MigrateMap.makeComputingMap(new Function<InstanceMode, InstanceConfigMonitor>() {

                public InstanceConfigMonitor apply(InstanceMode mode) {
-                    int scanInterval = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN_INTERVAL));
+                    int scanInterval = Integer
+                        .valueOf(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN_INTERVAL));

                    if (mode.isSpring()) {
                        SpringInstanceConfigMonitor monitor = new SpringInstanceConfigMonitor();
@@ -373,7 +377,8 @@ public class CanalController {
            InstanceConfig oldConfig = instanceConfigs.put(destination, config);

            if (oldConfig != null) {
-                logger.warn("destination:{} old config:{} has replace by new config:{}", destination, oldConfig, config);
+                logger
+                    .warn("destination:{} old config:{} has replace by new config:{}", destination, oldConfig, config);
            }
        }
    }
@@ -478,6 +483,7 @@ public class CanalController {
    }

    public void stop() throws Throwable {
+
        if (canalServer != null) {
            canalServer.stop();
        }
@@ -502,6 +508,19 @@ public class CanalController {

        if (zkclientx != null) {
            zkclientx.close();
+            if (zkServers != null) {
+                ZkClientx.removeZkClient(zkServers);
+            }
+        }
+
+        if (instanceConfigs != null) {
+            instanceConfigs.clear();
+        }
+        if (managerClients != null) {
+            managerClients.clear();
+        }
+        if (instanceConfigMonitors != null) {
+            instanceConfigMonitors.clear();
        }

        ZkClientx.clearClients();
--- a/meta/src/main/java/com/alibaba/otter/canal/meta/ZooKeeperMetaManager.java
+++ b/meta/src/main/java/com/alibaba/otter/canal/meta/ZooKeeperMetaManager.java
@@ -56,6 +56,7 @@ public class ZooKeeperMetaManager extends AbstractCanalLifeCycle implements Cana
    }

    public void stop() {
+        zkClientx = null;
        super.stop();
    }

@@ -101,6 +102,9 @@ public class ZooKeeperMetaManager extends AbstractCanalLifeCycle implements Cana
    }

    public List<ClientIdentity> listAllSubscribeInfo(String destination) throws CanalMetaManagerException {
+        if (zkClientx == null) {
+            return new ArrayList<ClientIdentity>();
+        }
        String path = ZookeeperPathUtils.getDestinationPath(destination);
        List<String> childs = null;
        try {
@@ -160,20 +164,19 @@ public class ZooKeeperMetaManager extends AbstractCanalLifeCycle implements Cana
    }

    public Long addBatch(ClientIdentity clientIdentity, PositionRange positionRange) throws CanalMetaManagerException {
-        String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(), clientIdentity.getClientId());
+        String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(),
+            clientIdentity.getClientId());
        byte[] data = JsonUtils.marshalToByte(positionRange, SerializerFeature.WriteClassName);
-        String batchPath = zkClientx.createPersistentSequential(path + ZookeeperPathUtils.ZOOKEEPER_SEPARATOR,
-            data,
-            true);
+        String batchPath = zkClientx
+            .createPersistentSequential(path + ZookeeperPathUtils.ZOOKEEPER_SEPARATOR, data, true);
        String batchIdString = StringUtils.substringAfterLast(batchPath, ZookeeperPathUtils.ZOOKEEPER_SEPARATOR);
        return ZookeeperPathUtils.getBatchMarkId(batchIdString);
    }

-    public void addBatch(ClientIdentity clientIdentity, PositionRange positionRange, Long batchId)
-                                                                                                  throws CanalMetaManagerException {
-        String path = ZookeeperPathUtils.getBatchMarkWithIdPath(clientIdentity.getDestination(),
-            clientIdentity.getClientId(),
-            batchId);
+    public void addBatch(ClientIdentity clientIdentity, PositionRange positionRange,
+                         Long batchId) throws CanalMetaManagerException {
+        String path = ZookeeperPathUtils
+            .getBatchMarkWithIdPath(clientIdentity.getDestination(), clientIdentity.getClientId(), batchId);
        byte[] data = JsonUtils.marshalToByte(positionRange, SerializerFeature.WriteClassName);
        zkClientx.createPersistent(path, data, true);
    }
@@ -204,9 +207,8 @@ public class ZooKeeperMetaManager extends AbstractCanalLifeCycle implements Cana
        }
        PositionRange positionRange = getBatch(clientIdentity, batchId);
        if (positionRange != null) {
-            String path = ZookeeperPathUtils.getBatchMarkWithIdPath(clientIdentity.getDestination(),
-                clientIdentity.getClientId(),
-                batchId);
+            String path = ZookeeperPathUtils
+                .getBatchMarkWithIdPath(clientIdentity.getDestination(), clientIdentity.getClientId(), batchId);
            zkClientx.delete(path);
        }

@@ -214,9 +216,8 @@ public class ZooKeeperMetaManager extends AbstractCanalLifeCycle implements Cana
    }

    public PositionRange getBatch(ClientIdentity clientIdentity, Long batchId) throws CanalMetaManagerException {
-        String path = ZookeeperPathUtils.getBatchMarkWithIdPath(clientIdentity.getDestination(),
-            clientIdentity.getClientId(),
-            batchId);
+        String path = ZookeeperPathUtils
+            .getBatchMarkWithIdPath(clientIdentity.getDestination(), clientIdentity.getClientId(), batchId);
        byte[] data = zkClientx.readData(path, true);
        if (data == null) {
            return null;
@@ -227,7 +228,8 @@ public class ZooKeeperMetaManager extends AbstractCanalLifeCycle implements Cana
    }

    public void clearAllBatchs(ClientIdentity clientIdentity) throws CanalMetaManagerException {
-        String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(), clientIdentity.getClientId());
+        String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(),
+            clientIdentity.getClientId());
        List<String> batchChilds = zkClientx.getChildren(path);

        for (String batchChild : batchChilds) {
@@ -237,7 +239,8 @@ public class ZooKeeperMetaManager extends AbstractCanalLifeCycle implements Cana
    }

    public PositionRange getLastestBatch(ClientIdentity clientIdentity) {
-        String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(), clientIdentity.getClientId());
+        String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(),
+            clientIdentity.getClientId());
        List<String> nodes = null;
        try {
            nodes = zkClientx.getChildren(path);
@@ -263,7 +266,8 @@ public class ZooKeeperMetaManager extends AbstractCanalLifeCycle implements Cana
    }

    public PositionRange getFirstBatch(ClientIdentity clientIdentity) {
-        String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(), clientIdentity.getClientId());
+        String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(),
+            clientIdentity.getClientId());
        List<String> nodes = null;
        try {
            nodes = zkClientx.getChildren(path);
@@ -289,7 +293,8 @@ public class ZooKeeperMetaManager extends AbstractCanalLifeCycle implements Cana
    }

    public Map<Long, PositionRange> listAllBatchs(ClientIdentity clientIdentity) {
-        String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(), clientIdentity.getClientId());
+        String path = ZookeeperPathUtils.getBatchMarkPath(clientIdentity.getDestination(),
+            clientIdentity.getClientId());
        List<String> nodes = null;
        try {
            nodes = zkClientx.getChildren(path);