fixed admin

canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/common/Threads.java

@@ -0,0 +1,17 @@
+package com.alibaba.otter.canal.admin.common;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+public class Threads {
+
+    public static int                   pool            = 60;
+    public static final ExecutorService executorService = new ThreadPoolExecutor(pool,
+                                                            pool,
+                                                            0L,
+                                                            TimeUnit.MILLISECONDS,
+                                                            new ArrayBlockingQueue<Runnable>(pool * 20),
+                                                            DaemonThreadFactory.daemonThreadFactory);
+}

canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/connector/AdminConnector.java

@@ -5,7 +5,7 @@ import com.alibaba.otter.canal.protocol.exception.CanalClientException;
 
 /**
  * canal数据操作客户端
- *
+ * 
  * @author zebin.xuzb @ 2012-6-19
  * @author jianghang
  * @version 1.0.0
@@ -14,14 +14,14 @@ public interface AdminConnector {
 
     /**
      * 链接对应的canal server
-     *
+     * 
      * @throws CanalClientException
      */
     void connect() throws ServiceException;
 
     /**
      * 释放链接
-     *
+     * 
      * @throws CanalClientException
      */
     void disconnect() throws ServiceException;
@@ -54,13 +54,6 @@ public interface AdminConnector {
      */
     boolean restart();
 
-    /**
-     * 获取所有当前节点下所有实例
-     *
-     * @return 实例信息
-     */
-    String getInstances();
-
     /**
      * 获取所有当前节点下运行中的实例
      *
@@ -70,7 +63,7 @@ public interface AdminConnector {
 
     /**
      * 通过实例名检查
-     *
+     * 
      * @param destination
      * @return
      */
@@ -92,6 +85,14 @@ public interface AdminConnector {
      */
     boolean stopInstance(String destination);
 
+    /**
+     * 通过实例名释放,主要针对cluster模式有效(通知当前主机释放instance运行交给其他人来抢占)
+     *
+     * @param destination 实例名
+     * @return 是否成功
+     */
+    boolean releaseInstance(String destination);
+
     /**
      * 通过实例名重启实例
      *
@@ -116,7 +117,7 @@ public interface AdminConnector {
 
     /**
      * 获取Instance的机器日志列表
-     *
+     * 
      * @param destination
      */
     String listInstanceLog(String destination);

canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/connector/SimpleAdminConnector.java

@@ -156,11 +156,6 @@ public class SimpleAdminConnector implements AdminConnector {
         return BooleanUtils.toBoolean(Integer.parseInt(doServerAdmin("restart")));
     }
 
-    @Override
-    public String getInstances() {
-        return doServerAdmin("instances");
-    }
-
     @Override
     public String getRunningInstances() {
         return doServerAdmin("list");
@@ -181,6 +176,11 @@ public class SimpleAdminConnector implements AdminConnector {
         return BooleanUtils.toBoolean(Integer.parseInt(doInstanceAdmin(destination, "stop")));
     }
 
+    @Override
+    public boolean releaseInstance(String destination) {
+        return BooleanUtils.toBoolean(Integer.parseInt(doInstanceAdmin(destination, "release")));
+    }
+
     @Override
     public boolean restartInstance(String destination) {
         return BooleanUtils.toBoolean(Integer.parseInt(doInstanceAdmin(destination, "restart")));
@@ -232,8 +232,11 @@ public class SimpleAdminConnector implements AdminConnector {
         try {
             writeWithHeader(Packet.newBuilder()
                 .setType(PacketType.INSTANCE)
-                .setBody(
-                    InstanceAdmin.newBuilder().setDestination(destination).setAction(action).build().toByteString())
+                .setBody(InstanceAdmin.newBuilder()
+                    .setDestination(destination)
+                    .setAction(action)
+                    .build()
+                    .toByteString())
                 .build()
                 .toByteArray());
 

canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/CanalInstanceController.java

@@ -162,6 +162,6 @@ public class CanalInstanceController {
      */
     @GetMapping(value = "/active/instances/{serverId}")
     public BaseModel<List<CanalInstanceConfig>> activeInstances(@PathVariable Long serverId, @PathVariable String env) {
-        return BaseModel.getInstance(canalInstanceConfigService.findActiveInstanceByServerId(serverId));
+        return BaseModel.getInstance(canalInstanceConfigService.findActiveInstaceByServerId(serverId));
     }
 }

canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/NodeServerController.java

@@ -1,9 +1,15 @@
 package com.alibaba.otter.canal.admin.controller;
 
-import java.util.List;
-
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.*;
+import org.springframework.web.bind.annotation.DeleteMapping;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.PutMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
 
 import com.alibaba.otter.canal.admin.model.BaseModel;
 import com.alibaba.otter.canal.admin.model.NodeServer;

canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/CanalInstanceService.java

@@ -20,6 +20,8 @@ public interface CanalInstanceService {
 
     CanalInstanceConfig detail(Long id);
 
+    CanalInstanceConfig findOne(String name);
+
     void updateContent(CanalInstanceConfig canalInstanceConfig);
 
     void delete(Long id);
@@ -30,5 +32,5 @@ public interface CanalInstanceService {
 
     boolean instanceOperation(Long id, String option);
 
-    List<CanalInstanceConfig> findActiveInstanceByServerId(Long serverId);
+    List<CanalInstanceConfig> findActiveInstaceByServerId(Long serverId);
 }

canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalClusterServiceImpl.java

@@ -1,17 +1,16 @@
 package com.alibaba.otter.canal.admin.service.impl;
 
+import io.ebean.Query;
+
 import java.util.List;
 
-import com.alibaba.otter.canal.admin.common.exception.ServiceException;
-import com.alibaba.otter.canal.admin.model.CanalInstanceConfig;
-import com.alibaba.otter.canal.admin.model.NodeServer;
 import org.springframework.stereotype.Service;
 
+import com.alibaba.otter.canal.admin.common.exception.ServiceException;
 import com.alibaba.otter.canal.admin.model.CanalCluster;
+import com.alibaba.otter.canal.admin.model.NodeServer;
 import com.alibaba.otter.canal.admin.service.CanalClusterServic;
 
-import io.ebean.Query;
-
 @Service
 public class CanalClusterServiceImpl implements CanalClusterServic {
 
@@ -31,16 +30,10 @@ public class CanalClusterServiceImpl implements CanalClusterServic {
         // 判断集群下是否存在server信息
         int serverCnt = NodeServer.find.query().where().eq("clusterId", id).findCount();
         if (serverCnt > 0) {
-            throw new ServiceException("Servers exist, delete failed");
-        }
-
-        // 判断集群下是否存在instance信息
-        int instanceCnt = CanalInstanceConfig.find.query().where().eq("clusterId", id).findCount();
-        if (instanceCnt > 0) {
-            throw new ServiceException("Instances exist, delete failed");
+            throw new ServiceException("当前集群下存在Server, 无法删除");
         }
 
-      CanalCluster canalCluster = CanalCluster.find.byId(id);
+        CanalCluster canalCluster = CanalCluster.find.byId(id);
         if (canalCluster != null) {
             canalCluster.delete();
         }

canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalConfigServiceImpl.java

@@ -5,19 +5,17 @@ import java.io.InputStream;
 import java.security.NoSuchAlgorithmException;
 import java.util.Date;
 
-import com.alibaba.otter.canal.admin.common.exception.ServiceException;
-import com.alibaba.otter.canal.admin.model.CanalCluster;
-import com.alibaba.otter.canal.admin.model.NodeServer;
-import com.alibaba.otter.canal.protocol.SecurityUtil;
-import io.ebean.Query;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
+import com.alibaba.otter.canal.admin.common.exception.ServiceException;
 import com.alibaba.otter.canal.admin.model.CanalConfig;
+import com.alibaba.otter.canal.admin.model.NodeServer;
 import com.alibaba.otter.canal.admin.service.CanalConfigService;
+import com.alibaba.otter.canal.protocol.SecurityUtil;
 
 /**
  * Canal配置信息业务层

canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalInstanceServiceImpl.java

@@ -1,23 +1,32 @@
 package com.alibaba.otter.canal.admin.service.impl;
 
+import io.ebean.Query;
+
 import java.security.NoSuchAlgorithmException;
-import java.util.*;
-import java.util.concurrent.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
-import com.alibaba.otter.canal.admin.model.Pager;
 import org.apache.commons.lang.StringUtils;
 import org.springframework.stereotype.Service;
 
-import com.alibaba.otter.canal.admin.common.DaemonThreadFactory;
+import com.alibaba.otter.canal.admin.common.Threads;
 import com.alibaba.otter.canal.admin.common.exception.ServiceException;
 import com.alibaba.otter.canal.admin.connector.AdminConnector;
 import com.alibaba.otter.canal.admin.connector.SimpleAdminConnectors;
 import com.alibaba.otter.canal.admin.model.CanalInstanceConfig;
 import com.alibaba.otter.canal.admin.model.NodeServer;
+import com.alibaba.otter.canal.admin.model.Pager;
 import com.alibaba.otter.canal.admin.service.CanalInstanceService;
 import com.alibaba.otter.canal.protocol.SecurityUtil;
-
-import io.ebean.Query;
+import com.google.common.collect.Lists;
 
 /**
  * Canal实例配置信息业务层
@@ -28,8 +37,7 @@ import io.ebean.Query;
 @Service
 public class CanalInstanceServiceImpl implements CanalInstanceService {
 
-    public Pager<CanalInstanceConfig> findList(CanalInstanceConfig canalInstanceConfig,
-                                               Pager<CanalInstanceConfig> pager) {
+    public Pager<CanalInstanceConfig> findList(CanalInstanceConfig canalInstanceConfig, Pager<CanalInstanceConfig> pager) {
         Query<CanalInstanceConfig> query = CanalInstanceConfig.find.query()
             .setDisableLazyLoading(true)
             .select("clusterId, serverId, name, modifiedTime")
@@ -62,12 +70,9 @@ public class CanalInstanceServiceImpl implements CanalInstanceService {
         }
 
         // check all canal instances running status
-        ExecutorService executorService = Executors.newFixedThreadPool(canalInstanceConfigs.size(),
-            DaemonThreadFactory.daemonThreadFactory);
         List<Future<Void>> futures = new ArrayList<>(canalInstanceConfigs.size());
-
         for (CanalInstanceConfig canalInstanceConfig1 : canalInstanceConfigs) {
-            futures.add(executorService.submit(() -> {
+            futures.add(Threads.executorService.submit(() -> {
                 List<NodeServer> nodeServers;
                 if (canalInstanceConfig1.getClusterId() != null) { // 集群模式
                     nodeServers = NodeServer.find.query()
@@ -81,15 +86,17 @@ public class CanalInstanceServiceImpl implements CanalInstanceService {
                 }
 
                 for (NodeServer nodeServer : nodeServers) {
-                    String runningInstances = SimpleAdminConnectors
-                        .execute(nodeServer.getIp(), nodeServer.getAdminPort(), AdminConnector::getRunningInstances);
+                    String runningInstances = SimpleAdminConnectors.execute(nodeServer.getIp(),
+                        nodeServer.getAdminPort(),
+                        AdminConnector::getRunningInstances);
                     if (runningInstances == null) {
                         continue;
                     }
                     String[] instances = runningInstances.split(",");
                     for (String instance : instances) {
                         if (instance.equals(canalInstanceConfig1.getName())) {
-                            if (canalInstanceConfig1.getNodeServer() == null) { // 集群模式下 server 对象为空
+                            // 集群模式下server对象为空
+                            if (canalInstanceConfig1.getNodeServer() == null) {
                                 canalInstanceConfig1.setNodeServer(nodeServer);
                             }
                             canalInstanceConfig1.setRunningStatus("1");
@@ -109,8 +116,6 @@ public class CanalInstanceServiceImpl implements CanalInstanceService {
                 // ignore
             }
         });
-        executorService.shutdownNow();
-
         return pager;
     }
 
@@ -119,29 +124,20 @@ public class CanalInstanceServiceImpl implements CanalInstanceService {
      *
      * @param serverId server id
      */
-    public List<CanalInstanceConfig> findActiveInstanceByServerId(Long serverId) {
+    public List<CanalInstanceConfig> findActiveInstaceByServerId(Long serverId) {
         NodeServer nodeServer = NodeServer.find.byId(serverId);
         if (nodeServer == null) {
             return null;
         }
-
-      String runningInstances = null;
-        if (nodeServer.getClusterId() != null) {// 集群模式
-            // 只取活动的instances
-            runningInstances = SimpleAdminConnectors
-                .execute(nodeServer.getIp(), nodeServer.getAdminPort(), AdminConnector::getRunningInstances);
-        } else {
-            // 取所属所有instances
-            runningInstances = SimpleAdminConnectors
-                .execute(nodeServer.getIp(), nodeServer.getAdminPort(), AdminConnector::getInstances);
-        }
-
+        String runningInstances = SimpleAdminConnectors.execute(nodeServer.getIp(),
+            nodeServer.getAdminPort(),
+            AdminConnector::getRunningInstances);
         if (runningInstances == null) {
             return null;
         }
 
         String[] instances = runningInstances.split(",");
-
+        Object obj[] = Lists.newArrayList(instances).toArray();
         // 单机模式和集群模式区分处理
         if (nodeServer.getClusterId() != null) { // 集群模式
             List<CanalInstanceConfig> list = CanalInstanceConfig.find.query()
@@ -149,7 +145,7 @@ public class CanalInstanceServiceImpl implements CanalInstanceService {
                 .select("clusterId, serverId, name, modifiedTime")
                 .where()
                 // 暂停的实例也显示 .eq("status", "1")
-                .in("name", instances)
+                .in("name", obj)
                 .findList();
             list.forEach(config -> config.setRunningStatus("1"));
             return list; // 集群模式直接返回当前运行的Instances
@@ -237,6 +233,16 @@ public class CanalInstanceServiceImpl implements CanalInstanceService {
         }
     }
 
+    @Override
+    public CanalInstanceConfig findOne(String name) {
+        CanalInstanceConfig config = CanalInstanceConfig.find.query()
+            .setDisableLazyLoading(true)
+            .where()
+            .eq("name", name)
+            .findOne();
+        return config;
+    }
+
     public Map<String, String> remoteInstanceLog(Long id, Long nodeId) {
         Map<String, String> result = new HashMap<>();
 
@@ -279,33 +285,19 @@ public class CanalInstanceServiceImpl implements CanalInstanceService {
         if (canalInstanceConfig == null) {
             return false;
         }
-        Boolean resutl = null;
+        Boolean result = null;
         if ("start".equals(option)) {
             if (nodeServer.getClusterId() == null) { // 非集群模式
                 return instanceOperation(id, "start");
-                // resutl = SimpleAdminConnectors.execute(nodeServer.getIp(),
-                // nodeServer.getAdminPort(),
-                // adminConnector ->
-                // adminConnector.startInstance(canalInstanceConfig.getName()));
+            } else {
+                throw new ServiceException("集群模式不允许指定server启动");
             }
         } else if ("stop".equals(option)) {
             if (nodeServer.getClusterId() != null) {
-                resutl = SimpleAdminConnectors.execute(nodeServer.getIp(),
+                // 集群模式,通知主动释放
+                result = SimpleAdminConnectors.execute(nodeServer.getIp(),
                     nodeServer.getAdminPort(),
-                    adminConnector -> adminConnector.stopInstance(canalInstanceConfig.getName()));
-
-                NodeServer nodeServerTmp = nodeServer;
-                // 集群模式下停止实例后过五秒钟再次启动进行启动抢占
-                Thread thread = new Thread(() -> {
-                    try {
-                        Thread.sleep(5000);
-                        SimpleAdminConnectors.execute(nodeServerTmp.getIp(),
-                            nodeServerTmp.getAdminPort(),
-                            adminConnector -> adminConnector.startInstance(canalInstanceConfig.getName()));
-                    } catch (Throwable e) {
-                    }
-                });
-                thread.start();
+                    adminConnector -> adminConnector.releaseInstance(canalInstanceConfig.getName()));
             } else { // 非集群模式下直接将状态置为0
                 return instanceOperation(id, "stop");
             }
@@ -313,10 +305,10 @@ public class CanalInstanceServiceImpl implements CanalInstanceService {
             return false;
         }
 
-        if (resutl == null) {
-            resutl = false;
+        if (result == null) {
+            result = false;
         }
-        return resutl;
+        return result;
     }
 
     public boolean instanceOperation(Long id, String option) {

canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/NodeServerServiceImpl.java

@@ -1,19 +1,25 @@
 package com.alibaba.otter.canal.admin.service.impl;
 
-import com.alibaba.otter.canal.admin.common.DaemonThreadFactory;
-import com.alibaba.otter.canal.admin.model.*;
 import io.ebean.Query;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.*;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import org.apache.commons.lang.StringUtils;
 import org.springframework.stereotype.Service;
 
+import com.alibaba.otter.canal.admin.common.Threads;
 import com.alibaba.otter.canal.admin.common.exception.ServiceException;
 import com.alibaba.otter.canal.admin.connector.AdminConnector;
 import com.alibaba.otter.canal.admin.connector.SimpleAdminConnectors;
+import com.alibaba.otter.canal.admin.model.CanalConfig;
+import com.alibaba.otter.canal.admin.model.CanalInstanceConfig;
+import com.alibaba.otter.canal.admin.model.NodeServer;
+import com.alibaba.otter.canal.admin.model.Pager;
 import com.alibaba.otter.canal.admin.service.NodeServerService;
 
 /**
@@ -123,12 +129,10 @@ public class NodeServerServiceImpl implements NodeServerService {
             return pager;
         }
 
-        ExecutorService executorService = Executors.newFixedThreadPool(nodeServers.size(),
-            DaemonThreadFactory.daemonThreadFactory);
         List<Future<Boolean>> futures = new ArrayList<>(nodeServers.size());
         // get all nodes status
         for (NodeServer ns : nodeServers) {
-            futures.add(executorService.submit(() -> {
+            futures.add(Threads.executorService.submit(() -> {
                 boolean status = SimpleAdminConnectors.execute(ns.getIp(), ns.getAdminPort(), AdminConnector::check);
                 ns.setStatus(status ? "1" : "0");
                 return !status;
@@ -142,8 +146,6 @@ public class NodeServerServiceImpl implements NodeServerService {
             }
         });
 
-        executorService.shutdownNow();
-
         return pager;
     }
 
@@ -157,8 +159,9 @@ public class NodeServerServiceImpl implements NodeServerService {
         if (nodeServer == null) {
             return "";
         }
-        return SimpleAdminConnectors
-            .execute(nodeServer.getIp(), nodeServer.getAdminPort(), adminConnector -> adminConnector.canalLog(100));
+        return SimpleAdminConnectors.execute(nodeServer.getIp(),
+            nodeServer.getAdminPort(),
+            adminConnector -> adminConnector.canalLog(100));
     }
 
     public boolean remoteOperation(Long id, String option) {
@@ -168,8 +171,7 @@ public class NodeServerServiceImpl implements NodeServerService {
         }
         Boolean result = null;
         if ("start".equals(option)) {
-            result = SimpleAdminConnectors
-                .execute(nodeServer.getIp(), nodeServer.getAdminPort(), AdminConnector::start);
+            result = SimpleAdminConnectors.execute(nodeServer.getIp(), nodeServer.getAdminPort(), AdminConnector::start);
         } else if ("stop".equals(option)) {
             result = SimpleAdminConnectors.execute(nodeServer.getIp(), nodeServer.getAdminPort(), AdminConnector::stop);
         } else {

canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/PollingConfigServiceImpl.java

@@ -7,6 +7,7 @@ import java.util.stream.Collectors;
 import org.apache.commons.lang.StringUtils;
 import org.springframework.stereotype.Service;
 
+import com.alibaba.otter.canal.admin.common.exception.ServiceException;
 import com.alibaba.otter.canal.admin.model.CanalConfig;
 import com.alibaba.otter.canal.admin.model.CanalInstanceConfig;
 import com.alibaba.otter.canal.admin.model.NodeServer;
@@ -28,7 +29,11 @@ public class PollingConfigServiceImpl implements PollingConfigServer {
         } else { // 单机模式
             canalConfig = CanalConfig.find.query().where().eq("serverId", server.getId()).findOne();
         }
-        if (canalConfig != null && !canalConfig.getContentMd5().equals(md5)) { // 内容发生变化
+        if (canalConfig == null) {
+            throw new ServiceException("canal.properties config is empty");
+        }
+
+        if (!canalConfig.getContentMd5().equals(md5)) { // 内容发生变化
             return canalConfig;
         }
         return null;

canal-admin/canal-admin-server/src/main/resources/application.yml

@@ -7,8 +7,8 @@ spring:
 
 spring.datasource:
   url: jdbc:mysql://127.0.0.1:3306/canal_manager?useUnicode=true&characterEncoding=UTF-8&useSSL=false
-  username: root
-  password: 121212
+  username: canal
+  password: canal
   driver-class-name: com.mysql.jdbc.Driver
   hikari:
     maximum-pool-size: 10

canal-admin/canal-admin-server/src/main/resources/public/static/css/chunk-1ef583f0.ec3a43f5.css

@@ -1 +0,0 @@
-.line[data-v-01afb045]{text-align:center}

canal-admin/canal-admin-server/src/main/resources/public/static/css/chunk-5560c396.b660603c.css

@@ -1 +0,0 @@
-.line[data-v-ae6e8604]{text-align:center}

canal-admin/canal-admin-server/src/main/resources/public/static/css/chunk-6f6d6145.bf63041a.css

@@ -1 +0,0 @@
-.line[data-v-250c9dcc]{text-align:center}

canal-admin/canal-admin-server/src/main/resources/public/static/js/chunk-238a81e9.273bda76.js

@@ -1 +0,0 @@
-(window["webpackJsonp"]=window["webpackJsonp"]||[]).push([["chunk-238a81e9"],{"26fc":function(t,s,a){t.exports=a.p+"static/img/404_cloud.0f4bc32b.png"},"8cdb":function(t,s,a){"use strict";a.r(s);var e=function(){var t=this,s=t.$createElement,a=t._self._c||s;return a("div",{staticClass:"wscn-http404-container"},[a("div",{staticClass:"wscn-http404"},[t._m(0),t._v(" "),a("div",{staticClass:"bullshit"},[a("div",{staticClass:"bullshit__oops"},[t._v("OOPS!")]),t._v(" "),t._m(1),t._v(" "),a("div",{staticClass:"bullshit__headline"},[t._v(t._s(t.message))]),t._v(" "),a("div",{staticClass:"bullshit__info"},[t._v("Please check that the URL you entered is correct, or click the button below to return to the homepage.")]),t._v(" "),a("a",{staticClass:"bullshit__return-home",attrs:{href:""}},[t._v("Back to home")])])])])},c=[function(){var t=this,s=t.$createElement,e=t._self._c||s;return e("div",{staticClass:"pic-404"},[e("img",{staticClass:"pic-404__parent",attrs:{src:a("a36b"),alt:"404"}}),t._v(" "),e("img",{staticClass:"pic-404__child left",attrs:{src:a("26fc"),alt:"404"}}),t._v(" "),e("img",{staticClass:"pic-404__child mid",attrs:{src:a("26fc"),alt:"404"}}),t._v(" "),e("img",{staticClass:"pic-404__child right",attrs:{src:a("26fc"),alt:"404"}})])},function(){var t=this,s=t.$createElement,a=t._self._c||s;return a("div",{staticClass:"bullshit__info"},[t._v("All rights reserved\n        "),a("a",{staticStyle:{color:"#20a0ff"},attrs:{href:"https://wallstreetcn.com",target:"_blank"}},[t._v("wallstreetcn")])])}],i={name:"Page404",computed:{message:function(){return"The webmaster said that you can not enter this page..."}}},l=i,n=(a("97ef"),a("2877")),r=Object(n["a"])(l,e,c,!1,null,"c095f994",null);s["default"]=r.exports},"97ef":function(t,s,a){"use strict";var e=a("b51e"),c=a.n(e);c.a},a36b:function(t,s,a){t.exports=a.p+"static/img/404.a57b6f31.png"},b51e:function(t,s,a){}}]);

canal-admin/canal-admin-server/src/main/resources/public/static/js/chunk-6f6d6145.8b7e1719.js

@@ -1 +0,0 @@
-(window["webpackJsonp"]=window["webpackJsonp"]||[]).push([["chunk-6f6d6145"],{"09a8":function(n,t,e){"use strict";e.r(t);var o=function(){var n=this,t=n.$createElement,e=n._self._c||t;return e("div",[e("el-form",{ref:"form",attrs:{model:n.form}},[e("div",{staticStyle:{"padding-left":"10px","padding-top":"20px"}},[e("el-form-item",[n._v("\n        "+n._s(n.form.name)+"    \n        "),e("el-button",{attrs:{type:"primary"},on:{click:n.onSubmit}},[n._v("修改")]),n._v(" "),e("el-button",{attrs:{type:"warning"},on:{click:n.onCancel}},[n._v("重置")])],1)],1),n._v(" "),e("editor",{attrs:{lang:"properties",theme:"chrome",width:"100%",height:800},on:{init:n.editorInit},model:{value:n.form.content,callback:function(t){n.$set(n.form,"content",t)},expression:"form.content"}})],1)],1)},a=[],c=(e("7f7f"),e("b775"));function i(){return Object(c["a"])({url:"/canal/config",method:"get"})}function r(n){return Object(c["a"])({url:"/canal/config",method:"put",data:n})}var f={components:{editor:e("7c9e")},data:function(){return{form:{id:null,name:"",content:""}}},created:function(){this.loadCanalConfig()},methods:{editorInit:function(){e("2099"),e("be9d"),e("2968"),e("e0e5"),e("bb36"),e("0329"),e("95b8"),e("6a21")},loadCanalConfig:function(){var n=this;i().then(function(t){var e=t.data;n.form.id=e.id,n.form.name=e.name,n.form.content=e.content})},onSubmit:function(){var n=this;this.$confirm("修改Server主配置可能会导致Server重启，是否继续？","确定修改",{confirmButtonText:"确定",cancelButtonText:"取消",type:"warning"}).then(function(){r(n.form).then(function(t){"success"===t.data?(n.$message({message:"修改成功",type:"success"}),n.loadCanalConfig()):n.$message({message:"修改失败",type:"error"})})})},onCancel:function(){this.loadCanalConfig()}}},s=f,u=(e("fbdc"),e("2877")),l=Object(u["a"])(s,o,a,!1,null,"250c9dcc",null);t["default"]=l.exports},8351:function(n,t,e){},fbdc:function(n,t,e){"use strict";var o=e("8351"),a=e.n(o);a.a}}]);

canal-admin/canal-admin-ui/src/views/canalServer/NodeServer.vue

@@ -72,7 +72,7 @@
               <el-dropdown-item @click.native="handleDelete(scope.row)">删除</el-dropdown-item>
               <el-dropdown-item @click.native="handleStart(scope.row)">启动</el-dropdown-item>
               <el-dropdown-item @click.native="handleStop(scope.row)">停止</el-dropdown-item>
-              <el-dropdown-item @click.native="handleInstances(scope.row)">实例</el-dropdown-item>
+              <el-dropdown-item @click.native="handleInstances(scope.row)">详情</el-dropdown-item>
               <el-dropdown-item @click.native="handleLog(scope.row)">日志</el-dropdown-item>
             </el-dropdown-menu>
           </el-dropdown>
@@ -113,7 +113,7 @@
         <el-button type="primary" @click="dataOperation()">确定</el-button>
       </div>
     </el-dialog>
-    <el-dialog :visible.sync="dialogInstances" title="实例列表" width="800px">
+    <el-dialog :visible.sync="dialogInstances" title="instance 列表" width="800px">
       <div class="filter-container">
         <el-button class="filter-item" type="info" @click="activeInstances()">刷新列表</el-button>
       </div>
@@ -310,6 +310,10 @@ export default {
       }
     },
     handleConfig(row) {
+      if (row.canalCluster !== null) {
+         this.$message({ message: '集群模式Server不允许单独变更配置，请在集群配置变更', type: 'error' })
+         return
+      }
       this.$router.push('/canalServer/nodeServer/config?serverId=' + row.id)
     },
     handleUpdate(row) {
@@ -374,7 +378,7 @@ export default {
         this.$message({ message: '当前Server不是启动状态，无法停止', type: 'error' })
         return
       }
-      this.$confirm('停止 Server 服务', '确定停止Server服务', {
+      this.$confirm('停止Server服务会导致所有Instance都停止服务', '确定停止Server服务', {
         confirmButtonText: '确定',
         cancelButtonText: '取消',
         type: 'warning'

common/src/main/java/com/alibaba/otter/canal/common/zookeeper/running/ServerRunningData.java

@@ -16,6 +16,7 @@ public class ServerRunningData implements Serializable {
 
     private static final long serialVersionUID = 92260481691855281L;
 
+    @Deprecated
     private Long              cid;
     private String            address;
     private boolean           active           = true;
@@ -23,8 +24,7 @@ public class ServerRunningData implements Serializable {
     public ServerRunningData(){
     }
 
-    public ServerRunningData(Long cid, String address){
-        this.cid = cid;
+    public ServerRunningData(String address){
         this.address = address;
     }
 

common/src/main/java/com/alibaba/otter/canal/common/zookeeper/running/ServerRunningMonitor.java

@@ -59,7 +59,6 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
                 }
 
                 if (!runningData.isActive() && isMine(runningData.getAddress())) { // 说明出现了主动释放的操作，并且本机之前是active
-                    release = true;
                     releaseRunning();// 彻底释放mainstem
                 }
 
@@ -116,7 +115,7 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
         if (zkClient != null) {
             releaseRunning(); // 尝试一下release
         } else {
-            processActiveExit(); // 没有zk，直接启动
+            processActiveExit(); // 没有zk，直接退出
         }
     }
 
@@ -185,8 +184,8 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
             boolean result = isMine(activeData.getAddress());
             if (!result) {
                 logger.warn("canal is running in node[{}] , but not in node[{}]",
-                    activeData.getCid(),
-                    serverData.getCid());
+                    activeData.getAddress(),
+                    serverData.getAddress());
             }
             return result;
         } catch (ZkNoNodeException e) {
@@ -204,6 +203,7 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
 
     private boolean releaseRunning() {
         if (check()) {
+            release = true;
             String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
             zkClient.delete(path);
             mutex.set(false);

common/src/test/java/com/alibaba/otter/canal/common/ServerRunningTest.java

@@ -16,6 +16,7 @@ import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
 import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningData;
 import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningListener;
 import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitor;
+
 @Ignore
 public class ServerRunningTest extends AbstractZkTest {
 
@@ -38,7 +39,7 @@ public class ServerRunningTest extends AbstractZkTest {
     @Test
     public void testOneServer() {
         final CountDownLatch countLatch = new CountDownLatch(2);
-        ServerRunningMonitor runningMonitor = buildServerRunning(countLatch, 1L, "127.0.0.1", 2088);
+        ServerRunningMonitor runningMonitor = buildServerRunning(countLatch, "127.0.0.1", 2088);
         runningMonitor.start();
         sleep(2000L);
         runningMonitor.stop();
@@ -52,9 +53,9 @@ public class ServerRunningTest extends AbstractZkTest {
     @Test
     public void testMultiServer() {
         final CountDownLatch countLatch = new CountDownLatch(30);
-        final ServerRunningMonitor runningMonitor1 = buildServerRunning(countLatch, 1L, "127.0.0.1", 2088);
-        final ServerRunningMonitor runningMonitor2 = buildServerRunning(countLatch, 2L, "127.0.0.1", 2089);
-        final ServerRunningMonitor runningMonitor3 = buildServerRunning(countLatch, 3L, "127.0.0.1", 2090);
+        final ServerRunningMonitor runningMonitor1 = buildServerRunning(countLatch, "127.0.0.1", 2088);
+        final ServerRunningMonitor runningMonitor2 = buildServerRunning(countLatch, "127.0.0.1", 2089);
+        final ServerRunningMonitor runningMonitor3 = buildServerRunning(countLatch, "127.0.0.1", 2090);
         final ExecutorService executor = Executors.newFixedThreadPool(3);
         executor.submit(new Runnable() {
 
@@ -110,29 +111,29 @@ public class ServerRunningTest extends AbstractZkTest {
         sleep(30000L);
     }
 
-    private ServerRunningMonitor buildServerRunning(final CountDownLatch countLatch, final Long cid, final String ip,
+    private ServerRunningMonitor buildServerRunning(final CountDownLatch countLatch, final final String ip,
                                                     final int port) {
-        ServerRunningData serverData = new ServerRunningData(cid, ip + ":" + port);
+        ServerRunningData serverData = new ServerRunningData(ip + ":" + port);
         ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData);
         runningMonitor.setDestination(destination);
         runningMonitor.setListener(new ServerRunningListener() {
 
             public void processActiveEnter() {
-                System.out.println(String.format("cid:%s ip:%s:%s has start", cid, ip, port));
+                System.out.println(String.format("cid:%s ip:%s:%s has start", ip, port));
                 countLatch.countDown();
             }
 
             public void processActiveExit() {
-                System.out.println(String.format("cid:%s ip:%s:%s has stop", cid, ip, port));
+                System.out.println(String.format("cid:%s ip:%s:%s has stop", ip, port));
                 countLatch.countDown();
             }
 
             public void processStart() {
-                System.out.println(String.format("cid:%s ip:%s:%s processStart", cid, ip, port));
+                System.out.println(String.format("cid:%s ip:%s:%s processStart", ip, port));
             }
 
             public void processStop() {
-                System.out.println(String.format("cid:%s ip:%s:%s processStop", cid, ip, port));
+                System.out.println(String.format("cid:%s ip:%s:%s processStop", ip, port));
             }
 
         });

deployer/src/main/bin/startup.bat

@@ -7,7 +7,8 @@ if "%OS%" == "Windows_NT" set ENV_PATH=%~dp0%
 
 set conf_dir=%ENV_PATH%\..\conf
 set canal_conf=%conf_dir%\canal.properties
-set logback_configurationFile=%conf_dir%\logback.xml
+set canal_local_conf=%conf_dir%\canal_local.properties
+@rem set logback_configurationFile=%conf_dir%\logback.xml
 
 set CLASSPATH=%conf_dir%
 set CLASSPATH=%conf_dir%\..\lib\*;%CLASSPATH%

deployer/src/main/bin/startup.sh

@@ -11,6 +11,7 @@ case "`uname`" in
 esac
 base=${bin_abs_path}/..
 canal_conf=$base/conf/canal.properties
+canal_local_conf=$base/conf/canal_local.properties
 logback_configurationFile=$base/conf/logback.xml
 export LANG=en_US.UTF-8
 export BASE=$base
@@ -48,21 +49,29 @@ in
 	;;
 1 )	
 	var=$*
-	if [ -f $var ] ; then 
-		canal_conf=$var
+	if [ "$var" = "local" ]; then
+		canal_conf=$canal_local_conf
 	else
-		echo "THE PARAMETER IS NOT CORRECT.PLEASE CHECK AGAIN."
-        exit
+		if [ -f $var ] ; then 
+			canal_conf=$var
+		else
+			echo "THE PARAMETER IS NOT CORRECT.PLEASE CHECK AGAIN."
+			exit
+		fi
 	fi;;
 2 )	
 	var=$1
-	if [ -f $var ] ; then
-		canal_conf=$var
-	else 
-		if [ "$1" = "debug" ]; then
-			DEBUG_PORT=$2
-			DEBUG_SUSPEND="n"
-			JAVA_DEBUG_OPT="-Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=$DEBUG_PORT,server=y,suspend=$DEBUG_SUSPEND"
+	if [ "$var" = "local" ]; then
+		canal_conf=$canal_local_conf
+	else
+		if [ -f $var ] ; then
+			canal_conf=$var
+		else 
+			if [ "$1" = "debug" ]; then
+				DEBUG_PORT=$2
+				DEBUG_SUSPEND="n"
+				JAVA_DEBUG_OPT="-Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=$DEBUG_PORT,server=y,suspend=$DEBUG_SUSPEND"
+			fi
 		fi
      fi;;
 * )

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java

@@ -46,7 +46,7 @@ import com.google.common.collect.MigrateMap;
  */
 public class CanalController {
 
-    private static final Logger                      logger           = LoggerFactory.getLogger(CanalController.class);
+    private static final Logger                      logger   = LoggerFactory.getLogger(CanalController.class);
     private Long                                     cid;
     private String                                   ip;
     private String                                   registerIp;
@@ -54,11 +54,10 @@ public class CanalController {
     private int                                      adminPort;
     // 默认使用spring的方式载入
     private Map<String, InstanceConfig>              instanceConfigs;
-    private Map<String, String>                      runningInstances = new MapMaker().makeMap();
     private InstanceConfig                           globalInstanceConfig;
     private Map<String, PlainCanalConfigClient>      managerClients;
     // 监听instance config的变化
-    private boolean                                  autoScan         = true;
+    private boolean                                  autoScan = true;
     private InstanceAction                           defaultAction;
     private Map<InstanceMode, InstanceConfigMonitor> instanceConfigMonitors;
     private CanalServerWithEmbedded                  embededCanalServer;
@@ -153,91 +152,88 @@ public class CanalController {
             zkclientx.createPersistent(ZookeeperPathUtils.CANAL_CLUSTER_ROOT_NODE, true);
         }
 
-        final ServerRunningData serverData = new ServerRunningData(cid, registerIp + ":" + port);
+        final ServerRunningData serverData = new ServerRunningData(registerIp + ":" + port);
         ServerRunningMonitors.setServerData(serverData);
-        ServerRunningMonitors
-            .setRunningMonitors(MigrateMap.makeComputingMap(new Function<String, ServerRunningMonitor>() {
-
-                public ServerRunningMonitor apply(final String destination) {
-                    ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData);
-                    runningMonitor.setDestination(destination);
-                    runningMonitor.setListener(new ServerRunningListener() {
-
-                        public void processActiveEnter() {
-                            try {
-                                MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
-                                embededCanalServer.start(destination);
-                                if (canalMQStarter != null) {
-                                    canalMQStarter.startDestination(destination);
-                                }
-                                runningInstances.put(destination, destination);
-                            } finally {
-                                MDC.remove(CanalConstants.MDC_DESTINATION);
+        ServerRunningMonitors.setRunningMonitors(MigrateMap.makeComputingMap(new Function<String, ServerRunningMonitor>() {
+
+            public ServerRunningMonitor apply(final String destination) {
+                ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData);
+                runningMonitor.setDestination(destination);
+                runningMonitor.setListener(new ServerRunningListener() {
+
+                    public void processActiveEnter() {
+                        try {
+                            MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
+                            embededCanalServer.start(destination);
+                            if (canalMQStarter != null) {
+                                canalMQStarter.startDestination(destination);
                             }
+                        } finally {
+                            MDC.remove(CanalConstants.MDC_DESTINATION);
                         }
+                    }
 
-                        public void processActiveExit() {
-                            try {
-                                MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
-                                if (canalMQStarter != null) {
-                                    canalMQStarter.stopDestination(destination);
-                                }
-                                embededCanalServer.stop(destination);
-                                runningInstances.remove(destination);
-                            } finally {
-                                MDC.remove(CanalConstants.MDC_DESTINATION);
+                    public void processActiveExit() {
+                        try {
+                            MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
+                            if (canalMQStarter != null) {
+                                canalMQStarter.stopDestination(destination);
                             }
+                            embededCanalServer.stop(destination);
+                        } finally {
+                            MDC.remove(CanalConstants.MDC_DESTINATION);
                         }
+                    }
+
+                    public void processStart() {
+                        try {
+                            if (zkclientx != null) {
+                                final String path = ZookeeperPathUtils.getDestinationClusterNode(destination,
+                                    registerIp + ":" + port);
+                                initCid(path);
+                                zkclientx.subscribeStateChanges(new IZkStateListener() {
+
+                                    public void handleStateChanged(KeeperState state) throws Exception {
+
+                                    }
+
+                                    public void handleNewSession() throws Exception {
+                                        initCid(path);
+                                    }
 
-                        public void processStart() {
-                            try {
-                                if (zkclientx != null) {
-                                    final String path = ZookeeperPathUtils.getDestinationClusterNode(destination,
-                                        registerIp + ":" + port);
-                                    initCid(path);
-                                    zkclientx.subscribeStateChanges(new IZkStateListener() {
-
-                                        public void handleStateChanged(KeeperState state) throws Exception {
-
-                                        }
-
-                                        public void handleNewSession() throws Exception {
-                                            initCid(path);
-                                        }
-
-                                        @Override
-                                        public void handleSessionEstablishmentError(Throwable error) throws Exception {
-                                            logger.error("failed to connect to zookeeper", error);
-                                        }
-                                    });
-                                }
-                            } finally {
-                                MDC.remove(CanalConstants.MDC_DESTINATION);
+                                    @Override
+                                    public void handleSessionEstablishmentError(Throwable error) throws Exception {
+                                        logger.error("failed to connect to zookeeper", error);
+                                    }
+                                });
                             }
+                        } finally {
+                            MDC.remove(CanalConstants.MDC_DESTINATION);
                         }
+                    }
 
-                        public void processStop() {
-                            try {
-                                MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
-                                if (zkclientx != null) {
-                                    final String path = ZookeeperPathUtils.getDestinationClusterNode(destination,
-                                        registerIp + ":" + port);
-                                    releaseCid(path);
-                                }
-                            } finally {
-                                MDC.remove(CanalConstants.MDC_DESTINATION);
+                    public void processStop() {
+                        try {
+                            MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
+                            if (zkclientx != null) {
+                                final String path = ZookeeperPathUtils.getDestinationClusterNode(destination,
+                                    registerIp + ":" + port);
+                                releaseCid(path);
                             }
+                        } finally {
+                            MDC.remove(CanalConstants.MDC_DESTINATION);
                         }
-
-                    });
-                    if (zkclientx != null) {
-                        runningMonitor.setZkClient(zkclientx);
                     }
-                    // 触发创建一下cid节点
-                    runningMonitor.init();
-                    return runningMonitor;
+
+                });
+                if (zkclientx != null) {
+                    runningMonitor.setZkClient(zkclientx);
                 }
-            }));
+                // 触发创建一下cid节点
+                runningMonitor.init();
+                return runningMonitor;
+            }
+        }));
 
         // 初始化monitor机制
         autoScan = BooleanUtils.toBoolean(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN));
@@ -284,13 +280,26 @@ public class CanalController {
 
                     logger.info("auto notify reload {} successful.", destination);
                 }
+
+                @Override
+                public void release(String destination) {
+                    // 此处的release，代表强制释放，主要针对HA机制释放运行，让给其他机器抢占
+                    InstanceConfig config = instanceConfigs.get(destination);
+                    if (config != null) {
+                        ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
+                        if (runningMonitor.isStart()) {
+                            runningMonitor.release();
+                        }
+                    }
+
+                    logger.info("auto notify release {} successful.", destination);
+                }
             };
 
             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();
@@ -366,7 +375,7 @@ public class CanalController {
                 }
 
                 if (config.getMode().isManager()) {
-                    PlainCanalInstanceGenerator instanceGenerator = new PlainCanalInstanceGenerator(properties);
+                    PlainCanalInstanceGenerator instanceGenerator = new PlainCanalInstanceGenerator();
                     instanceGenerator.setCanalConfigClient(managerClients.get(config.getManagerAddress()));
                     instanceGenerator.setSpringXml(config.getSpringXml());
                     return instanceGenerator.generate(destination);
@@ -386,7 +395,7 @@ public class CanalController {
     }
 
     private PlainCanalConfigClient getManagerClient(String managerAddress) {
-        return new PlainCanalConfigClient(managerAddress, this.adminUser, this.adminPasswd, null, adminPort);
+        return new PlainCanalConfigClient(managerAddress, this.adminUser, this.adminPasswd, this.registerIp, adminPort);
     }
 
     private void initInstanceConfig(Properties properties) {
@@ -398,8 +407,7 @@ 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);
             }
         }
     }
@@ -595,7 +603,4 @@ public class CanalController {
         return instanceConfigs;
     }
 
-    public Map<String, String> getRunningInstances() {
-        return runningInstances;
-    }
 }

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java

@@ -11,6 +11,7 @@ import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.alibaba.otter.canal.common.utils.AddressUtils;
 import com.alibaba.otter.canal.common.utils.NamedThreadFactory;
 import com.alibaba.otter.canal.instance.manager.plain.PlainCanal;
 import com.alibaba.otter.canal.instance.manager.plain.PlainCanalConfigClient;
@@ -27,7 +28,7 @@ public class CanalLauncher {
     private static final Logger             logger               = LoggerFactory.getLogger(CanalLauncher.class);
     public static final CountDownLatch      runningLatch         = new CountDownLatch(1);
     private static ScheduledExecutorService executor             = Executors.newScheduledThreadPool(1,
-        new NamedThreadFactory("canal-server-scan"));
+                                                                     new NamedThreadFactory("canal-server-scan"));
 
     public static void main(String[] args) {
         try {
@@ -49,55 +50,54 @@ public class CanalLauncher {
             if (StringUtils.isNotEmpty(managerAddress)) {
                 String user = properties.getProperty(CanalConstants.CANAL_ADMIN_USER);
                 String passwd = properties.getProperty(CanalConstants.CANAL_ADMIN_PASSWD);
-                String adminPort = properties.getProperty(CanalConstants.CANAL_ADMIN_PORT, "11110");
+                String adminPort = properties.getProperty(CanalConstants.CANAL_ADMIN_PORT);
+                String registerIp = properties.getProperty(CanalConstants.CANAL_REGISTER_IP);
+                if (StringUtils.isEmpty(adminPort)) {
+                    adminPort = "11110";
+                }
+                if (StringUtils.isEmpty(registerIp)) {
+                    registerIp = AddressUtils.getHostIp();
+                }
                 final PlainCanalConfigClient configClient = new PlainCanalConfigClient(managerAddress,
                     user,
                     passwd,
-                    "",
+                    registerIp,
                     Integer.parseInt(adminPort));
                 PlainCanal canalConfig = configClient.findServer(null);
-                if (canalConfig != null) {
-                    properties = canalConfig.getProperties();
-                    // 用本地配置覆盖
-                    properties.put(CanalConstants.CANAL_ADMIN_MANAGER, managerAddress);
-                    properties.put(CanalConstants.CANAL_ADMIN_USER, user);
-                    properties.put(CanalConstants.CANAL_ADMIN_PASSWD, passwd);
-                    properties.put(CanalConstants.CANAL_ADMIN_PORT, adminPort);
-                    int scanIntervalInSecond = Integer
-                        .parseInt(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL, "5"));
-                    executor.scheduleWithFixedDelay(new Runnable() {
-
-                        private PlainCanal lastCanalConfig;
-
-                        public void run() {
-                            try {
-                                if (lastCanalConfig == null) {
-                                    lastCanalConfig = configClient.findServer(null);
-                                } else {
-                                    PlainCanal newCanalConfig = configClient.findServer(lastCanalConfig.getMd5());
-                                    if (newCanalConfig != null) {
-                                        // 远程配置canal.properties修改重新加载整个应用
-                                        canalStater.stop();
-                                        Properties properties1 = newCanalConfig.getProperties();
-                                        // 用本地配置覆盖
-                                        properties1.put(CanalConstants.CANAL_ADMIN_MANAGER, managerAddress);
-                                        properties1.put(CanalConstants.CANAL_ADMIN_USER, user);
-                                        properties1.put(CanalConstants.CANAL_ADMIN_PASSWD, passwd);
-                                        properties1.put(CanalConstants.CANAL_ADMIN_PORT, adminPort);
-                                        canalStater.setProperties(properties1);
-                                        canalStater.start();
-
-                                        lastCanalConfig = newCanalConfig;
-                                    }
+                if (canalConfig == null) {
+                    throw new IllegalArgumentException("managerAddress:" + managerAddress
+                                                       + " can't not found config for [" + registerIp + ":" + adminPort
+                                                       + "]");
+                }
+                properties = canalConfig.getProperties();
+                int scanIntervalInSecond = Integer.valueOf(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL,
+                    "5"));
+                executor.scheduleWithFixedDelay(new Runnable() {
+
+                    private PlainCanal lastCanalConfig;
+
+                    public void run() {
+                        try {
+                            if (lastCanalConfig == null) {
+                                lastCanalConfig = configClient.findServer(null);
+                            } else {
+                                PlainCanal newCanalConfig = configClient.findServer(lastCanalConfig.getMd5());
+                                if (newCanalConfig != null) {
+                                    // 远程配置canal.properties修改重新加载整个应用
+                                    canalStater.stop();
+                                    canalStater.setProperties(newCanalConfig.getProperties());
+                                    canalStater.start();
+
+                                    lastCanalConfig = newCanalConfig;
                                 }
-
-                            } catch (Throwable e) {
-                                logger.error("scan failed", e);
                             }
+
+                        } catch (Throwable e) {
+                            logger.error("scan failed", e);
                         }
+                    }
 
-                    }, 0, scanIntervalInSecond, TimeUnit.SECONDS);
-                }
+                }, 0, scanIntervalInSecond, TimeUnit.SECONDS);
             }
 
             canalStater.setProperties(properties);
@@ -118,4 +118,5 @@ public class CanalLauncher {
             }
         });
     }
+
 }

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java

@@ -1,9 +1,5 @@
 package com.alibaba.otter.canal.deployer;
 
-import java.io.File;
-import java.io.FileFilter;
-import java.util.Arrays;
-import java.util.List;
 import java.util.Properties;
 
 import org.apache.commons.lang.StringUtils;
@@ -17,9 +13,6 @@ import com.alibaba.otter.canal.kafka.CanalKafkaProducer;
 import com.alibaba.otter.canal.rocketmq.CanalRocketMQProducer;
 import com.alibaba.otter.canal.server.CanalMQStarter;
 import com.alibaba.otter.canal.spi.CanalMQProducer;
-import com.google.common.base.Function;
-import com.google.common.base.Joiner;
-import com.google.common.collect.Lists;
 
 /**
  * Canal server 启动类
@@ -76,40 +69,8 @@ public class CanalStater {
         if (canalMQProducer != null) {
             // disable netty
             System.setProperty(CanalConstants.CANAL_WITHOUT_NETTY, "true");
-            String autoScan = CanalController.getProperty(properties, CanalConstants.CANAL_AUTO_SCAN);
             // 设置为raw避免ByteString->Entry的二次解析
             System.setProperty("canal.instance.memory.rawEntry", "false");
-            if ("true".equals(autoScan)) {
-                String rootDir = CanalController.getProperty(properties, CanalConstants.CANAL_CONF_DIR);
-                if (StringUtils.isEmpty(rootDir)) {
-                    rootDir = "../conf";
-                }
-                File rootdir = new File(rootDir);
-                if (rootdir.exists()) {
-                    File[] instanceDirs = rootdir.listFiles(new FileFilter() {
-
-                        public boolean accept(File pathname) {
-                            String filename = pathname.getName();
-                            return pathname.isDirectory() && !"spring".equalsIgnoreCase(filename)
-                                   && !"metrics".equalsIgnoreCase(filename);
-                        }
-                    });
-                    if (instanceDirs != null && instanceDirs.length > 0) {
-                        List<String> instances = Lists.transform(Arrays.asList(instanceDirs),
-                            new Function<File, String>() {
-
-                                @Override
-                                public String apply(File instanceDir) {
-                                    return instanceDir.getName();
-                                }
-                            });
-                        System.setProperty(CanalConstants.CANAL_DESTINATIONS, Joiner.on(",").join(instances));
-                    }
-                }
-            } else {
-                String destinations = CanalController.getProperty(properties, CanalConstants.CANAL_DESTINATIONS);
-                System.setProperty(CanalConstants.CANAL_DESTINATIONS, destinations);
-            }
         }
 
         logger.info("## start the canal server.");
@@ -136,7 +97,8 @@ public class CanalStater {
         if (canalMQProducer != null) {
             canalMQStarter = new CanalMQStarter(canalMQProducer);
             MQProperties mqProperties = buildMQProperties(properties);
-            canalMQStarter.start(mqProperties);
+            String destinations = CanalController.getProperty(properties, CanalConstants.CANAL_DESTINATIONS);
+            canalMQStarter.start(mqProperties, destinations);
             controller.setCanalMQStarter(canalMQStarter);
         }
 

deployer/src/main/java/com/alibaba/otter/canal/deployer/admin/CanalAdminController.java

@@ -2,6 +2,7 @@ package com.alibaba.otter.canal.deployer.admin;
 
 import java.io.File;
 import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
@@ -14,7 +15,6 @@ import org.slf4j.LoggerFactory;
 
 import com.alibaba.otter.canal.admin.CanalAdmin;
 import com.alibaba.otter.canal.common.utils.FileUtils;
-import com.alibaba.otter.canal.deployer.CanalController;
 import com.alibaba.otter.canal.deployer.CanalStater;
 import com.alibaba.otter.canal.deployer.InstanceConfig;
 import com.alibaba.otter.canal.deployer.monitor.InstanceAction;
@@ -28,7 +28,7 @@ import com.google.common.base.Joiner;
 
 /**
  * 提供canal admin的管理操作
- *
+ * 
  * @author agapple 2019年8月24日 下午11:39:01
  * @since 1.1.4
  */
@@ -102,32 +102,18 @@ public class CanalAdminController implements CanalAdmin {
         return false;
     }
 
-    @Override
-    public String getInstances() {
-        try {
-            CanalController controller = canalStater.getController();
-            if (controller != null) {
-                Map<String, InstanceConfig> instanceConfigs = controller.getInstanceConfigs();
-                if (instanceConfigs != null) {
-                    return Joiner.on(",").join(instanceConfigs.keySet());
-                }
-            }
-        } catch (Throwable e) {
-            logger.error(e.getMessage(), e);
-        }
-        return "";
-    }
-
     @Override
     public String getRunningInstances() {
         try {
-            CanalController controller = canalStater.getController();
-            if (controller != null) {
-                Map<String, String> instanceConfigs = controller.getRunningInstances();
-                if (instanceConfigs != null) {
-                    return Joiner.on(",").join(instanceConfigs.keySet());
+            Map<String, CanalInstance> instances = CanalServerWithEmbedded.instance().getCanalInstances();
+            List<String> runningInstances = new ArrayList<String>();
+            instances.forEach((destination, instance) -> {
+                if (instance.isStart()) {
+                    runningInstances.add(destination);
                 }
-            }
+            });
+
+            return Joiner.on(",").join(runningInstances);
         } catch (Throwable e) {
             logger.error(e.getMessage(), e);
         }
@@ -173,6 +159,20 @@ public class CanalAdminController implements CanalAdmin {
         return false;
     }
 
+    @Override
+    public boolean releaseInstance(String destination) {
+        try {
+            InstanceAction instanceAction = getInstanceAction(destination);
+            if (instanceAction != null) {
+                instanceAction.release(destination);
+                return true;
+            }
+        } catch (Throwable e) {
+            logger.error(e.getMessage(), e);
+        }
+        return false;
+    }
+
     @Override
     public boolean restartInstance(String destination) {
         try {

deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/InstanceAction.java

@@ -13,6 +13,11 @@ public interface InstanceAction {
      */
     void start(String destination);
 
+    /**
+     * 主动释放destination运行
+     */
+    void release(String destination);
+
     /**
      * 停止destination
      */

deployer/src/main/resources/canal.properties

@@ -1,7 +1,6 @@
 #################################################
 ######### 		common argument		#############
 #################################################
-canal.id = 1
 # tcp bind ip
 canal.ip =
 # register ip to zookeeper
@@ -12,12 +11,6 @@ canal.metrics.pull.port = 11112
 # canal.user = canal
 # canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458
 
-# canal admin config
-#canal.admin.manager = 127.0.0.1:8089
-canal.admin.port = 11110
-canal.admin.user = admin
-canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
-
 canal.zkServers =
 # flush data to zk
 canal.zookeeper.flush.period = 1000

deployer/src/main/resources/canal_local.properties

@@ -0,0 +1,8 @@
+# register ip
+canal.register.ip =
+
+# canal admin config
+canal.admin.manager = 127.0.0.1:8089
+canal.admin.port = 11110
+canal.admin.user = admin
+canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441

deployer/src/main/resources/canal_manager.properties

@@ -0,0 +1,139 @@
+#################################################
+######### 		common argument		#############
+#################################################
+# tcp bind ip
+canal.ip =
+# register ip to zookeeper
+canal.register.ip =
+canal.port = 11111
+canal.metrics.pull.port = 11112
+# canal instance user/passwd
+canal.user = canal
+canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458
+
+# canal admin config
+canal.admin.manager = 127.0.0.1:8089
+canal.admin.port = 11110
+canal.admin.user = admin
+canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
+
+canal.zkServers =
+# flush data to zk
+canal.zookeeper.flush.period = 1000
+canal.withoutNetty = false
+# tcp, kafka, RocketMQ
+canal.serverMode = tcp
+# flush meta cursor/parse position to file
+canal.file.data.dir = ${canal.conf.dir}
+canal.file.flush.period = 1000
+## memory store RingBuffer size, should be Math.pow(2,n)
+canal.instance.memory.buffer.size = 16384
+## memory store RingBuffer used memory unit size , default 1kb
+canal.instance.memory.buffer.memunit = 1024 
+## meory store gets mode used MEMSIZE or ITEMSIZE
+canal.instance.memory.batch.mode = MEMSIZE
+canal.instance.memory.rawEntry = true
+
+## detecing config
+canal.instance.detecting.enable = false
+#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
+canal.instance.detecting.sql = select 1
+canal.instance.detecting.interval.time = 3
+canal.instance.detecting.retry.threshold = 3
+canal.instance.detecting.heartbeatHaEnable = false
+
+# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
+canal.instance.transaction.size =  1024
+# mysql fallback connected to new master should fallback times
+canal.instance.fallbackIntervalInSeconds = 60
+
+# network config
+canal.instance.network.receiveBufferSize = 16384
+canal.instance.network.sendBufferSize = 16384
+canal.instance.network.soTimeout = 30
+
+# binlog filter config
+canal.instance.filter.druid.ddl = true
+canal.instance.filter.query.dcl = false
+canal.instance.filter.query.dml = false
+canal.instance.filter.query.ddl = false
+canal.instance.filter.table.error = false
+canal.instance.filter.rows = false
+canal.instance.filter.transaction.entry = false
+
+# binlog format/image check
+canal.instance.binlog.format = ROW,STATEMENT,MIXED 
+canal.instance.binlog.image = FULL,MINIMAL,NOBLOB
+
+# binlog ddl isolation
+canal.instance.get.ddl.isolation = false
+
+# parallel parser config
+canal.instance.parser.parallel = true
+## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
+#canal.instance.parser.parallelThreadSize = 16
+## disruptor ringbuffer size, must be power of 2
+canal.instance.parser.parallelBufferSize = 256
+
+# table meta tsdb info
+canal.instance.tsdb.enable = true
+canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
+canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
+canal.instance.tsdb.dbUsername = canal
+canal.instance.tsdb.dbPassword = canal
+# dump snapshot interval, default 24 hour
+canal.instance.tsdb.snapshot.interval = 24
+# purge snapshot expire , default 360 hour(15 days)
+canal.instance.tsdb.snapshot.expire = 360
+
+# aliyun ak/sk , support rds/mq
+canal.aliyun.accessKey =
+canal.aliyun.secretKey =
+
+#################################################
+######### 		destinations		#############
+#################################################
+canal.destinations = 
+# conf root dir
+canal.conf.dir = ../conf
+# auto scan instance dir add/remove and start/stop instance
+canal.auto.scan = true
+canal.auto.scan.interval = 5
+
+canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
+#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
+
+canal.instance.global.mode = manager
+canal.instance.global.lazy = false
+canal.instance.global.manager.address = ${canal.admin.manager}
+#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
+canal.instance.global.spring.xml = classpath:spring/file-instance.xml
+#canal.instance.global.spring.xml = classpath:spring/default-instance.xml
+
+##################################################
+######### 		     MQ 		     #############
+##################################################
+canal.mq.servers = 127.0.0.1:6667
+canal.mq.retries = 0
+canal.mq.batchSize = 16384
+canal.mq.maxRequestSize = 1048576
+canal.mq.lingerMs = 100
+canal.mq.bufferMemory = 33554432
+canal.mq.canalBatchSize = 50
+canal.mq.canalGetTimeout = 100
+canal.mq.flatMessage = true
+canal.mq.compressionType = none
+canal.mq.acks = all
+#canal.mq.properties. =
+canal.mq.producerGroup = test
+# Set this value to "cloud", if you want open message trace feature in aliyun.
+canal.mq.accessChannel = local
+# aliyun mq namespace
+#canal.mq.namespace =
+
+##################################################
+#########     Kafka Kerberos Info    #############
+##################################################
+canal.mq.kafka.kerberos.enable = false
+canal.mq.kafka.kerberos.krb5FilePath = "../conf/kerberos/krb5.conf"
+canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf"

instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/PlainCanalInstanceGenerator.java

@@ -13,11 +13,9 @@ import com.alibaba.otter.canal.instance.manager.plain.PlainCanal;
 import com.alibaba.otter.canal.instance.manager.plain.PlainCanalConfigClient;
 import com.alibaba.otter.canal.instance.spring.SpringCanalInstanceGenerator;
 
-import java.util.Properties;
-
 /**
  * 基于manager生成对应的{@linkplain CanalInstance}
- *
+ * 
  * @author jianghang 2012-7-12 下午05:37:09
  * @version 1.0.0
  */
@@ -28,11 +26,6 @@ public class PlainCanalInstanceGenerator implements CanalInstanceGenerator {
     private PlainCanalConfigClient canalConfigClient;
     private String                 defaultName = "instance";
     private BeanFactory            beanFactory;
-    private Properties             canalConfig;
-
-    public PlainCanalInstanceGenerator(Properties canalConfig){
-        this.canalConfig = canalConfig;
-    }
 
     public CanalInstance generate(String destination) {
         synchronized (CanalInstanceGenerator.class) {
@@ -41,11 +34,8 @@ public class PlainCanalInstanceGenerator implements CanalInstanceGenerator {
                 if (canal == null) {
                     throw new CanalException("instance : " + destination + " config is not found");
                 }
-                Properties properties = canal.getProperties();
-                properties.putAll(canalConfig);
                 // 设置动态properties,替换掉本地properties
-                com.alibaba.otter.canal.instance.spring.support.PropertyPlaceholderConfigurer.propertiesLocal
-                    .set(properties);
+                com.alibaba.otter.canal.instance.spring.support.PropertyPlaceholderConfigurer.propertiesLocal.set(canal.getProperties());
                 // 设置当前正在加载的通道，加载spring查找文件时会用到该变量
                 System.setProperty("canal.instance.destination", destination);
                 this.beanFactory = getBeanFactory(springXml);

instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanalConfigClient.java

@@ -2,8 +2,6 @@ package com.alibaba.otter.canal.instance.manager.plain;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
 import java.nio.charset.StandardCharsets;
 import java.security.NoSuchAlgorithmException;
 import java.util.HashMap;
@@ -46,14 +44,10 @@ public class PlainCanalConfigClient extends AbstractCanalLifeCycle implements Ca
         this.user = user;
         this.passwd = passwd;
         this.httpHelper = new HttpHelper();
-        if (StringUtils.isNotEmpty(localIp)) {
-            this.localIp = localIp;
+        if (StringUtils.isEmpty(localIp)) {
+            this.localIp = "127.0.0.1";// 本地测试用
         } else {
-            try {
-                this.localIp = InetAddress.getLocalHost().getHostAddress();
-            } catch (UnknownHostException e) {
-                e.printStackTrace();
-            }
+            this.localIp = localIp;
         }
         this.adminPort = adminPort;
     }

server/src/main/java/com/alibaba/otter/canal/admin/CanalAdmin.java

@@ -2,7 +2,7 @@ package com.alibaba.otter.canal.admin;
 
 /**
  * Canal Admin动态管理接口
- *
+ * 
  * @author agapple 2019年8月24日 下午9:45:49
  * @since 1.1.4
  */
@@ -41,13 +41,6 @@ public interface CanalAdmin {
      */
     boolean restart();
 
-    /**
-     * 获取所有当前节点下所有实例
-     *
-     * @return 实例信息
-     */
-    String getInstances();
-
     /**
      * 获取所有当前节点下运行中的实例
      *
@@ -57,7 +50,7 @@ public interface CanalAdmin {
 
     /**
      * 通过实例名检查
-     *
+     * 
      * @param destination
      * @return
      */
@@ -79,6 +72,14 @@ public interface CanalAdmin {
      */
     boolean stopInstance(String destination);
 
+    /**
+     * 通过实例名释放,主要针对cluster模式有效(通知当前主机释放instance运行交给其他人来抢占)
+     *
+     * @param destination 实例名
+     * @return 是否成功
+     */
+    boolean releaseInstance(String destination);
+
     /**
      * 通过实例名重启实例
      *
@@ -103,7 +104,7 @@ public interface CanalAdmin {
 
     /**
      * 获取Instance的机器日志列表
-     *
+     * 
      * @param destination
      */
     String listInstanceLog(String destination);

server/src/main/java/com/alibaba/otter/canal/admin/handler/SessionHandler.java

@@ -58,9 +58,6 @@ public class SessionHandler extends SimpleChannelHandler {
                         case "list":
                             message = canalAdmin.getRunningInstances();
                             break;
-                        case "instances":
-                            message = canalAdmin.getInstances();
-                            break;
                         default:
                             byte[] errorBytes = AdminNettyUtils.errorPacket(301,
                                 MessageFormatter.format("ServerAdmin action={} is unknown", action).getMessage());
@@ -83,6 +80,9 @@ public class SessionHandler extends SimpleChannelHandler {
                         case "stop":
                             message = canalAdmin.stopInstance(destination) ? "1" : "0";
                             break;
+                        case "release":
+                            message = canalAdmin.releaseInstance(destination) ? "1" : "0";
+                            break;
                         case "restart":
                             message = canalAdmin.restartInstance(destination) ? "1" : "0";
                             break;
@@ -132,11 +132,9 @@ public class SessionHandler extends SimpleChannelHandler {
             }
         } catch (Throwable exception) {
             byte[] errorBytes = AdminNettyUtils.errorPacket(400,
-                MessageFormatter
-                    .format("something goes wrong with channel:{}, exception={}",
-                        ctx.getChannel(),
-                        ExceptionUtils.getStackTrace(exception))
-                    .getMessage());
+                MessageFormatter.format("something goes wrong with channel:{}, exception={}",
+                    ctx.getChannel(),
+                    ExceptionUtils.getStackTrace(exception)).getMessage());
             AdminNettyUtils.write(ctx.getChannel(), errorBytes);
         }
     }

server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java

@@ -42,7 +42,7 @@ public class CanalMQStarter {
         this.canalMQProducer = canalMQProducer;
     }
 
-    public synchronized void start(MQProperties properties) {
+    public synchronized void start(MQProperties properties, String destinations) {
         try {
             if (running) {
                 return;
@@ -60,8 +60,8 @@ public class CanalMQStarter {
             executorService = Executors.newCachedThreadPool();
             logger.info("## start the MQ workers.");
 
-            String[] destinations = StringUtils.split(System.getProperty("canal.destinations"), ",");
-            for (String destination : destinations) {
+            String[] dsts = StringUtils.split(destinations, ",");
+            for (String destination : dsts) {
                 destination = destination.trim();
                 CanalMQRunnable canalMQRunnable = new CanalMQRunnable(destination);
                 canalMQWorks.put(destination, canalMQRunnable);