canal集成

smtweb-framework/canal/client/canal.example.iml

@@ -27,7 +27,7 @@
     </content>
     <orderEntry type="inheritedJdk" />
     <orderEntry type="sourceFolder" forTests="false" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.client:1.1.6" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.client:1.1.5" level="project" />
     <orderEntry type="library" name="Maven: com.google.protobuf:protobuf-java:3.6.1" level="project" />
     <orderEntry type="library" name="Maven: io.netty:netty-all:4.1.78.Final" level="project" />
     <orderEntry type="library" name="Maven: io.netty:netty-buffer:4.1.78.Final" level="project" />
@@ -68,6 +68,7 @@
     <orderEntry type="library" name="Maven: commons-io:commons-io:2.4" level="project" />
     <orderEntry type="library" name="Maven: commons-lang:commons-lang:2.6" level="project" />
     <orderEntry type="library" name="Maven: commons-codec:commons-codec:1.15" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba:fastjson:1.2.58.sec06" level="project" />
     <orderEntry type="library" name="Maven: com.google.guava:guava:22.0" level="project" />
     <orderEntry type="library" name="Maven: com.google.code.findbugs:jsr305:1.3.9" level="project" />
     <orderEntry type="library" name="Maven: com.google.errorprone:error_prone_annotations:2.0.18" level="project" />
@@ -86,10 +87,9 @@
     <orderEntry type="library" name="Maven: org.springframework:spring-jdbc:5.3.21" level="project" />
     <orderEntry type="library" name="Maven: org.springframework:spring-tx:5.3.21" level="project" />
     <orderEntry type="library" name="Maven: org.springframework:spring-orm:5.3.21" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.protocol:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.common:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.fastjson2:fastjson2:2.0.4" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba:druid:1.2.11" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.protocol:1.1.5" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.common:1.1.5" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba:druid:1.2.6" level="project" />
     <orderEntry type="module" module-name="sw-system-bpm" />
     <orderEntry type="library" name="Maven: org.springframework.boot:spring-boot-starter-web:2.6.9" level="project" />
     <orderEntry type="library" name="Maven: org.springframework.boot:spring-boot-starter:2.6.9" level="project" />

smtweb-framework/canal/client/pom.xml

@@ -9,23 +9,23 @@
 
 	<groupId>cc.smtweb</groupId>
 	<artifactId>canal.example</artifactId>
-	<version>1.1.6</version>
+	<version>1.1.5</version>
 
 	<dependencies>
 		<dependency>
 			<groupId>com.alibaba.otter</groupId>
 			<artifactId>canal.client</artifactId>
-			<version>1.1.6</version>
+			<version>1.1.5</version>
 		</dependency>
 		<dependency>
 			<groupId>com.alibaba.otter</groupId>
 			<artifactId>canal.protocol</artifactId>
-			<version>1.1.6</version>
+			<version>1.1.5</version>
 		</dependency>
 		<dependency>
 			<groupId>com.alibaba</groupId>
 			<artifactId>druid</artifactId>
-			<version>1.2.11</version>
+			<version>1.2.6</version>
 		</dependency>
 
 		<dependency>
@@ -118,9 +118,9 @@
 		<plugins>
 			<!-- deploy模块的packaging通常是jar，如果项目中没有java 源代码或资源文件，加上这一段配置使项目能通过构建 -->
 			<plugin>
-                <groupId>org.apache.maven.plugins</groupId>
+				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-jar-plugin</artifactId>
-                <version>2.5</version>
+				<version>2.5</version>
 				<configuration>
 					<archive>
 						<addMavenDescriptor>true</addMavenDescriptor>
@@ -198,7 +198,7 @@
 								<descriptor>${basedir}/src/main/assembly/release.xml</descriptor>
 							</descriptors>
 							<!-- 如果一个应用的包含多个deploy模块，如果使用同样的包名， 如果把它们复制的一个目录中可能会失败，所以包名加了 artifactId以示区分 -->
-							<finalName>${project.artifactId}-1.1.6</finalName>
+							<finalName>${project.artifactId}-1.1.5</finalName>
 							<!-- scm 要求 release 模式打出的包放到顶级目录下的target子目录中 -->
 							<outputDirectory>${project.parent.build.directory}</outputDirectory>
 						</configuration>

smtweb-framework/canal/client/src/main/java/cc/smtweb/system/canal/example/ClientConsts.java

@@ -12,7 +12,7 @@ import java.util.Properties;
  */
 public class ClientConsts {
 
-    public static final String ROOT                                 = "canal.client";
+    public static final String ROOT                                 = "canal.server";
     public static final String SIZE                                 = ROOT + "." + "size";
     public static final String INSTANCE                             = ROOT + "." + "instance";
     public static final String IP                                   = ROOT + "." + "ip";

smtweb-framework/canal/client/src/main/java/cc/smtweb/system/canal/example/ClientInstance.java

@@ -165,9 +165,9 @@ public class ClientInstance {
                     logger.error("canal服务连接异常！", "重新连接canal服务端已失败【" + count + "】次，请检查canal服务是否启动，请及时启动canal服务端，或联系研发人员！");
                 }
                 logger.debug("canal服务连接中断，第【" + count + "】次重连失败：" + CommUtil.getOrigMsg(e2));
-                logger.debug("canal服务连接中断，" + (3 * rate) + "分钟后进行第【" + (count + 1) + "】连接测试！");
+                logger.debug("canal服务连接中断，" + (1 * rate) + "分钟后进行第【" + (count + 1) + "】连接测试！");
                 try {
-                    Thread.sleep(180 * rate * 1000L);
+                    Thread.sleep(60 * rate * 1000L);
                 } catch (InterruptedException e1) {
                     // ignore
                 }

smtweb-framework/canal/client/src/main/resources/client.properties

@@ -9,6 +9,6 @@ canal.server.port=11111
 canal.server.username=
 canal.server.password=
 # 数据库匹配规则.*\\..*, scmz\\..*
-canal.server.filter=.*\\..*
+canal.server.filter=scmz\\..*
 
 

smtweb-framework/canal/server/canal.deployer.iml

@@ -27,9 +27,9 @@
     </content>
     <orderEntry type="inheritedJdk" />
     <orderEntry type="sourceFolder" forTests="false" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.server:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.instance.core:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.common:1.1.6" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.server:1.1.5" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.instance.core:1.1.5" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.common:1.1.5" level="project" />
     <orderEntry type="library" name="Maven: io.netty:netty-all:4.1.78.Final" level="project" />
     <orderEntry type="library" name="Maven: io.netty:netty-codec-dns:4.1.78.Final" level="project" />
     <orderEntry type="library" name="Maven: io.netty:netty-codec-haproxy:4.1.78.Final" level="project" />
@@ -60,7 +60,7 @@
     <orderEntry type="library" name="Maven: org.jboss.netty:netty:3.2.2.Final" level="project" />
     <orderEntry type="library" name="Maven: com.101tec:zkclient:0.10" level="project" />
     <orderEntry type="library" name="Maven: commons-lang:commons-lang:2.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.fastjson2:fastjson2:2.0.4" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba:fastjson:1.2.58.sec06" level="project" />
     <orderEntry type="library" name="Maven: com.google.guava:guava:22.0" level="project" />
     <orderEntry type="library" name="Maven: com.google.j2objc:j2objc-annotations:1.1" level="project" />
     <orderEntry type="library" name="Maven: org.codehaus.mojo:animal-sniffer-annotations:1.14" level="project" />
@@ -68,33 +68,34 @@
     <orderEntry type="library" name="Maven: ch.qos.logback:logback-classic:1.2.11" level="project" />
     <orderEntry type="library" name="Maven: org.slf4j:jcl-over-slf4j:1.7.36" level="project" />
     <orderEntry type="library" name="Maven: org.springframework:spring-aop:5.3.21" level="project" />
+    <orderEntry type="library" name="Maven: org.springframework:spring-context:5.3.21" level="project" />
     <orderEntry type="library" name="Maven: org.springframework:spring-orm:5.3.21" level="project" />
     <orderEntry type="library" name="Maven: com.google.code.findbugs:jsr305:3.0.2" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.store:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.protocol:1.1.6" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.store:1.1.5" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.protocol:1.1.5" level="project" />
     <orderEntry type="library" name="Maven: com.google.protobuf:protobuf-java:3.6.1" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.meta:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.parse:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.sink:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.parse.dbsync:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.filter:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.parse.driver:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba:druid:1.2.10" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.meta:1.1.5" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.parse:1.1.5" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.sink:1.1.5" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.parse.dbsync:1.1.5" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.filter:1.1.5" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.parse.driver:1.1.5" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba:druid:1.2.6" level="project" />
     <orderEntry type="library" name="Maven: org.mybatis:mybatis-spring:2.0.4" level="project" />
     <orderEntry type="library" name="Maven: org.mybatis:mybatis:3.5.4" level="project" />
     <orderEntry type="library" name="Maven: com.h2database:h2:1.4.200" level="project" />
     <orderEntry type="library" name="Maven: org.apache.commons:commons-compress:1.9" level="project" />
     <orderEntry type="library" name="Maven: com.lmax:disruptor:3.4.2" level="project" />
     <orderEntry type="library" name="Maven: javax.annotation:javax.annotation-api:1.3.2" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.instance.spring:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.instance.manager:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: com.alibaba.otter:connector.core:1.1.6" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.instance.spring:1.1.5" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:canal.instance.manager:1.1.5" level="project" />
+    <orderEntry type="library" name="Maven: com.alibaba.otter:connector.core:1.1.5" level="project" />
     <orderEntry type="library" name="Maven: com.googlecode.aviator:aviator:2.2.1" level="project" />
     <orderEntry type="library" name="Maven: commons-beanutils:commons-beanutils:1.8.2" level="project" />
     <orderEntry type="library" name="Maven: oro:oro:2.0.8" level="project" />
     <orderEntry type="library" name="Maven: joda-time:joda-time:2.9.4" level="project" />
     <orderEntry type="library" name="Maven: org.apache.commons:commons-lang3:3.12.0" level="project" />
-    <orderEntry type="library" scope="RUNTIME" name="Maven: com.alibaba.otter:canal.prometheus:1.1.6" level="project" />
+    <orderEntry type="library" scope="RUNTIME" name="Maven: com.alibaba.otter:canal.prometheus:1.1.5" level="project" />
     <orderEntry type="library" scope="RUNTIME" name="Maven: org.jctools:jctools-core:2.1.2" level="project" />
     <orderEntry type="library" scope="RUNTIME" name="Maven: io.prometheus:simpleclient:0.12.0" level="project" />
     <orderEntry type="library" scope="RUNTIME" name="Maven: io.prometheus:simpleclient_tracer_otel:0.12.0" level="project" />
@@ -106,24 +107,20 @@
     <orderEntry type="library" scope="RUNTIME" name="Maven: io.prometheus:simpleclient_pushgateway:0.12.0" level="project" />
     <orderEntry type="library" name="Maven: javax.xml.bind:jaxb-api:2.3.1" level="project" />
     <orderEntry type="library" name="Maven: javax.activation:javax.activation-api:1.2.0" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.alibaba.otter:connector.kafka:jar-with-dependencies:1.1.6" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.alibaba.otter:connector.rocketmq:jar-with-dependencies:1.1.6" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.alibaba.otter:connector.rabbitmq:jar-with-dependencies:1.1.6" level="project" />
-    <orderEntry type="library" scope="PROVIDED" name="Maven: com.alibaba.otter:connector.pulsarmq:jar-with-dependencies:1.1.6" level="project" />
-    <orderEntry type="library" name="Maven: org.springframework.boot:spring-boot-autoconfigure:2.6.9" level="project" />
-    <orderEntry type="library" name="Maven: org.springframework.boot:spring-boot:2.6.9" level="project" />
-    <orderEntry type="library" name="Maven: org.springframework:spring-core:5.3.21" level="project" />
-    <orderEntry type="library" name="Maven: org.springframework:spring-jcl:5.3.21" level="project" />
-    <orderEntry type="library" name="Maven: org.springframework:spring-context:5.3.21" level="project" />
-    <orderEntry type="library" name="Maven: org.springframework:spring-beans:5.3.21" level="project" />
-    <orderEntry type="library" name="Maven: org.springframework:spring-expression:5.3.21" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.alibaba.otter:connector.kafka:jar-with-dependencies:1.1.5" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.alibaba.otter:connector.rocketmq:jar-with-dependencies:1.1.5" level="project" />
+    <orderEntry type="library" scope="PROVIDED" name="Maven: com.alibaba.otter:connector.rabbitmq:jar-with-dependencies:1.1.5" level="project" />
     <orderEntry type="module" module-name="sw-framework-core" />
     <orderEntry type="library" name="Maven: org.springframework.boot:spring-boot-starter:2.6.9" level="project" />
+    <orderEntry type="library" name="Maven: org.springframework.boot:spring-boot:2.6.9" level="project" />
+    <orderEntry type="library" name="Maven: org.springframework.boot:spring-boot-autoconfigure:2.6.9" level="project" />
     <orderEntry type="library" name="Maven: org.springframework.boot:spring-boot-starter-logging:2.6.9" level="project" />
     <orderEntry type="library" name="Maven: org.apache.logging.log4j:log4j-to-slf4j:2.17.2" level="project" />
     <orderEntry type="library" name="Maven: org.apache.logging.log4j:log4j-api:2.17.2" level="project" />
     <orderEntry type="library" name="Maven: org.slf4j:jul-to-slf4j:1.7.36" level="project" />
     <orderEntry type="library" name="Maven: jakarta.annotation:jakarta.annotation-api:1.3.5" level="project" />
+    <orderEntry type="library" name="Maven: org.springframework:spring-core:5.3.21" level="project" />
+    <orderEntry type="library" name="Maven: org.springframework:spring-jcl:5.3.21" level="project" />
     <orderEntry type="library" name="Maven: org.yaml:snakeyaml:1.29" level="project" />
     <orderEntry type="library" name="Maven: org.springframework.boot:spring-boot-starter-web:2.6.9" level="project" />
     <orderEntry type="library" name="Maven: org.springframework.boot:spring-boot-starter-json:2.6.9" level="project" />
@@ -135,7 +132,9 @@
     <orderEntry type="library" name="Maven: org.apache.tomcat.embed:tomcat-embed-el:9.0.64" level="project" />
     <orderEntry type="library" name="Maven: org.apache.tomcat.embed:tomcat-embed-websocket:9.0.64" level="project" />
     <orderEntry type="library" name="Maven: org.springframework:spring-web:5.3.21" level="project" />
+    <orderEntry type="library" name="Maven: org.springframework:spring-beans:5.3.21" level="project" />
     <orderEntry type="library" name="Maven: org.springframework:spring-webmvc:5.3.21" level="project" />
+    <orderEntry type="library" name="Maven: org.springframework:spring-expression:5.3.21" level="project" />
     <orderEntry type="library" name="Maven: org.apache.httpcomponents:httpclient:4.5.13" level="project" />
     <orderEntry type="library" name="Maven: org.apache.httpcomponents:httpcore:4.4.15" level="project" />
     <orderEntry type="library" name="Maven: com.fasterxml.jackson.core:jackson-core:2.13.3" level="project" />

smtweb-framework/canal/server/pom.xml

@@ -11,27 +11,27 @@
 
     <groupId>cc.smtweb</groupId>
     <artifactId>canal.deployer</artifactId>
-    <version>1.1.6</version>
+    <version>1.1.5</version>
 
     <dependencies>
-
         <dependency>
             <groupId>com.alibaba.otter</groupId>
             <artifactId>canal.server</artifactId>
-            <version>1.1.6</version>
+            <version>1.1.5</version>
         </dependency>
+
         <!-- 这里指定runtime的metrics provider-->
         <dependency>
             <groupId>com.alibaba.otter</groupId>
             <artifactId>canal.prometheus</artifactId>
-            <version>1.1.6</version>
+            <version>1.1.5</version>
             <scope>runtime</scope>
         </dependency>
 
         <dependency>
             <groupId>com.alibaba.otter</groupId>
             <artifactId>connector.kafka</artifactId>
-            <version>1.1.6</version>
+            <version>1.1.5</version>
             <exclusions>
                 <exclusion>
                     <artifactId>*</artifactId>
@@ -44,7 +44,7 @@
         <dependency>
             <groupId>com.alibaba.otter</groupId>
             <artifactId>connector.rocketmq</artifactId>
-            <version>1.1.6</version>
+            <version>1.1.5</version>
             <exclusions>
                 <exclusion>
                     <artifactId>*</artifactId>
@@ -57,20 +57,7 @@
         <dependency>
             <groupId>com.alibaba.otter</groupId>
             <artifactId>connector.rabbitmq</artifactId>
-            <version>1.1.6</version>
-            <exclusions>
-                <exclusion>
-                    <artifactId>*</artifactId>
-                    <groupId>*</groupId>
-                </exclusion>
-            </exclusions>
-            <classifier>jar-with-dependencies</classifier>
-            <scope>provided</scope>
-        </dependency>
-        <dependency>
-            <groupId>com.alibaba.otter</groupId>
-            <artifactId>connector.pulsarmq</artifactId>
-            <version>1.1.6</version>
+            <version>1.1.5</version>
             <exclusions>
                 <exclusion>
                     <artifactId>*</artifactId>
@@ -82,11 +69,6 @@
         </dependency>
 
         <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-autoconfigure</artifactId>
-        </dependency>
-
-        <dependency>
             <groupId>cc.smtweb</groupId>
             <artifactId>sw-framework-core</artifactId>
             <version>3.1.0-SNAPSHOT</version>
@@ -97,16 +79,11 @@
             <artifactId>sw-system-bpm</artifactId>
             <version>3.1.0-SNAPSHOT</version>
         </dependency>
+
     </dependencies>
 
     <build>
         <plugins>
-
-            <!--<plugin>
-                <groupId>org.springframework.boot</groupId>
-                <artifactId>spring-boot-maven-plugin</artifactId>
-            </plugin>-->
-
             <plugin>
                 <artifactId>maven-jar-plugin</artifactId>
                 <configuration>
@@ -210,7 +187,7 @@
                                 <descriptor>${basedir}/src/main/assembly/release.xml</descriptor>
                             </descriptors>
                             <!-- 如果一个应用的包含多个deploy模块，如果使用同样的包名， 如果把它们复制的一个目录中可能会失败，所以包名加了 artifactId以示区分 -->
-                            <finalName>${project.artifactId}-1.1.6</finalName>
+                            <finalName>${project.artifactId}-1.1.5</finalName>
                             <!-- scm 要求 release 模式打出的包放到顶级目录下的target子目录中 -->
                             <outputDirectory>${project.parent.build.directory}</outputDirectory>
                         </configuration>

smtweb-framework/canal/server/src/main/java/cc/smtweb/system/canal/deployer/CanalController.java

@@ -1,22 +1,9 @@
 package cc.smtweb.system.canal.deployer;
 
-import java.util.Map;
-import java.util.Properties;
-
 import cc.smtweb.system.canal.deployer.monitor.InstanceAction;
 import cc.smtweb.system.canal.deployer.monitor.InstanceConfigMonitor;
 import cc.smtweb.system.canal.deployer.monitor.ManagerInstanceConfigMonitor;
 import cc.smtweb.system.canal.deployer.monitor.SpringInstanceConfigMonitor;
-import org.I0Itec.zkclient.IZkStateListener;
-import org.I0Itec.zkclient.exception.ZkNoNodeException;
-import org.I0Itec.zkclient.exception.ZkNodeExistsException;
-import org.apache.commons.lang.BooleanUtils;
-import org.apache.commons.lang.StringUtils;
-import org.apache.zookeeper.Watcher.Event.KeeperState;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.slf4j.MDC;
-
 import com.alibaba.otter.canal.common.utils.AddressUtils;
 import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
 import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
@@ -24,7 +11,6 @@ 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;
 import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitors;
-import cc.smtweb.system.canal.deployer.InstanceConfig.InstanceMode;
 import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator;
 import com.alibaba.otter.canal.instance.manager.PlainCanalInstanceGenerator;
 import com.alibaba.otter.canal.instance.manager.plain.PlainCanalConfigClient;
@@ -36,6 +22,18 @@ import com.alibaba.otter.canal.server.netty.CanalServerWithNetty;
 import com.google.common.base.Function;
 import com.google.common.collect.MapMaker;
 import com.google.common.collect.MigrateMap;
+import org.I0Itec.zkclient.IZkStateListener;
+import org.I0Itec.zkclient.exception.ZkNoNodeException;
+import org.I0Itec.zkclient.exception.ZkNodeExistsException;
+import org.apache.commons.lang.BooleanUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+
+import java.util.Map;
+import java.util.Properties;
 
 /**
  * canal调度控制器
@@ -57,7 +55,7 @@ public class CanalController {
     // 监听instance config的变化
     private boolean                                  autoScan = true;
     private InstanceAction defaultAction;
-    private Map<InstanceMode, InstanceConfigMonitor> instanceConfigMonitors;
+    private Map<InstanceConfig.InstanceMode, InstanceConfigMonitor> instanceConfigMonitors;
     private CanalServerWithEmbedded                  embededCanalServer;
     private CanalServerWithNetty                     canalServer;
 
@@ -336,9 +334,9 @@ public class CanalController {
         String modeStr = getProperty(properties, CanalConstants.getInstanceModeKey(CanalConstants.GLOBAL_NAME));
         if (StringUtils.isNotEmpty(adminManagerAddress)) {
             // 如果指定了manager地址,则强制适用manager
-            globalConfig.setMode(InstanceMode.MANAGER);
+            globalConfig.setMode(InstanceConfig.InstanceMode.MANAGER);
         } else if (StringUtils.isNotEmpty(modeStr)) {
-            globalConfig.setMode(InstanceMode.valueOf(StringUtils.upperCase(modeStr)));
+            globalConfig.setMode(InstanceConfig.InstanceMode.valueOf(StringUtils.upperCase(modeStr)));
         }
 
         String lazyStr = getProperty(properties, CanalConstants.getInstancLazyKey(CanalConstants.GLOBAL_NAME));
@@ -409,9 +407,9 @@ public class CanalController {
         String modeStr = getProperty(properties, CanalConstants.getInstanceModeKey(destination));
         if (StringUtils.isNotEmpty(adminManagerAddress)) {
             // 如果指定了manager地址,则强制适用manager
-            config.setMode(InstanceMode.MANAGER);
+            config.setMode(InstanceConfig.InstanceMode.MANAGER);
         } else if (StringUtils.isNotEmpty(modeStr)) {
-            config.setMode(InstanceMode.valueOf(StringUtils.upperCase(modeStr)));
+            config.setMode(InstanceConfig.InstanceMode.valueOf(StringUtils.upperCase(modeStr)));
         }
 
         String lazyStr = getProperty(properties, CanalConstants.getInstancLazyKey(destination));
@@ -595,7 +593,7 @@ public class CanalController {
         this.canalMQStarter = canalMQStarter;
     }
 
-    public Map<InstanceMode, InstanceConfigMonitor> getInstanceConfigMonitors() {
+    public Map<InstanceConfig.InstanceMode, InstanceConfigMonitor> getInstanceConfigMonitors() {
         return instanceConfigMonitors;
     }
 

smtweb-framework/canal/server/src/main/java/cc/smtweb/system/canal/deployer/CanalLauncher.java

@@ -1,14 +1,5 @@
 package cc.smtweb.system.canal.deployer;
 
-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;
-import org.apache.commons.lang.BooleanUtils;
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.FileInputStream;
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
@@ -16,6 +7,16 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.commons.lang.BooleanUtils;
+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;
+
 /**
  * canal独立版本启动的入口类
  *
@@ -24,25 +25,21 @@ import java.util.concurrent.TimeUnit;
  */
 public class CanalLauncher {
 
-    private static final String CLASSPATH_URL_PREFIX = "classpath:";
-    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"));
+    private static final String             CLASSPATH_URL_PREFIX = "classpath:";
+    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"));
 
     public static void main(String[] args) {
         startServer();
     }
 
-    public static void startServer() {
+    public static void startServer(){
         try {
-            logger.info("## canal sever start begin");
             logger.info("## set default uncaught exception handler");
             setGlobalUncaughtExceptionHandler();
 
-            // 支持rocketmq client 配置日志路径
-            System.setProperty("rocketmq.client.logUseSlf4j", "true");
-
             logger.info("## load canal configurations");
             String conf = System.getProperty("canal.conf", "classpath:canal.properties");
             Properties properties = new Properties();
@@ -63,10 +60,6 @@ public class CanalLauncher {
                         CanalConstants.CANAL_ADMIN_AUTO_REGISTER));
                 String autoCluster = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_AUTO_CLUSTER);
                 String name = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_REGISTER_NAME);
-                if (StringUtils.isEmpty(name)) {
-                    name = AddressUtils.getHostName();
-                }
-
                 String registerIp = CanalController.getProperty(properties, CanalConstants.CANAL_REGISTER_IP);
                 if (StringUtils.isEmpty(registerIp)) {
                     registerIp = AddressUtils.getHostIp();
@@ -126,7 +119,6 @@ public class CanalLauncher {
             }
 
             canalStater.start();
-            logger.info("## canal server start success");
             runningLatch.await();
             executor.shutdownNow();
         } catch (Throwable e) {

smtweb-framework/canal/server/src/main/java/cc/smtweb/system/canal/deployer/CanalStarter.java

@@ -1,17 +1,16 @@
 package cc.smtweb.system.canal.deployer;
 
-import java.util.Properties;
-
+import cc.smtweb.system.canal.deployer.admin.CanalAdminController;
+import com.alibaba.otter.canal.admin.netty.CanalAdminWithNetty;
 import com.alibaba.otter.canal.connector.core.config.MQProperties;
+import com.alibaba.otter.canal.connector.core.spi.CanalMQProducer;
+import com.alibaba.otter.canal.connector.core.spi.ExtensionLoader;
+import com.alibaba.otter.canal.server.CanalMQStarter;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.alibaba.otter.canal.admin.netty.CanalAdminWithNetty;
-import com.alibaba.otter.canal.connector.core.spi.CanalMQProducer;
-import com.alibaba.otter.canal.connector.core.spi.ExtensionLoader;
-import cc.smtweb.system.canal.deployer.admin.CanalAdminController;
-import com.alibaba.otter.canal.server.CanalMQStarter;
+import java.util.Properties;
 
 /**
  * Canal server 启动类

smtweb-framework/canal/server/src/main/java/cc/smtweb/system/canal/deployer/admin/CanalAdminController.java

@@ -1,30 +1,29 @@
 package cc.smtweb.system.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;
-import java.util.stream.Collectors;
-
 import cc.smtweb.system.canal.deployer.CanalStarter;
+import cc.smtweb.system.canal.deployer.InstanceConfig;
+import cc.smtweb.system.canal.deployer.monitor.InstanceAction;
 import cc.smtweb.system.canal.deployer.monitor.InstanceConfigMonitor;
 import cc.smtweb.system.canal.deployer.monitor.ManagerInstanceConfigMonitor;
 import cc.smtweb.system.canal.deployer.monitor.SpringInstanceConfigMonitor;
-import cc.smtweb.system.canal.deployer.monitor.InstanceAction;
-import org.apache.commons.io.filefilter.TrueFileFilter;
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.alibaba.otter.canal.admin.CanalAdmin;
 import com.alibaba.otter.canal.common.utils.FileUtils;
-import cc.smtweb.system.canal.deployer.InstanceConfig;
 import com.alibaba.otter.canal.instance.core.CanalInstance;
 import com.alibaba.otter.canal.protocol.SecurityUtil;
 import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
 import com.google.common.base.Joiner;
+import org.apache.commons.io.filefilter.TrueFileFilter;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * 提供canal admin的管理操作
@@ -37,7 +36,7 @@ public class CanalAdminController implements CanalAdmin {
     private static final Logger logger = LoggerFactory.getLogger(CanalAdminController.class);
     private String              user;
     private String              passwd;
-    private CanalStarter canalStater;
+    private CanalStarter         canalStater;
 
     public CanalAdminController(CanalStarter canalStater){
         this.canalStater = canalStater;

smtweb-framework/canal/server/src/main/java/cc/smtweb/system/canal/deployer/monitor/ManagerInstanceConfigMonitor.java

@@ -1,6 +1,5 @@
 package cc.smtweb.system.canal.deployer.monitor;
 
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Executors;
@@ -78,9 +77,9 @@ public class ManagerInstanceConfigMonitor extends AbstractCanalLifeCycle impleme
         }
 
         final List<String> is = Lists.newArrayList(StringUtils.split(instances, ','));
-        List<String> start = new ArrayList<>();
-        List<String> stop = new ArrayList<>();
-        List<String> restart = new ArrayList<>();
+        List<String> start = Lists.newArrayList();
+        List<String> stop = Lists.newArrayList();
+        List<String> restart = Lists.newArrayList();
         for (String instance : is) {
             if (!configs.containsKey(instance)) {
                 PlainCanal newPlainCanal = configClient.findInstance(instance, null);

smtweb-framework/canal/server/src/main/resources/logback.xml.bak → smtweb-framework/canal/server/src/main/resources/logback.xml

@@ -79,39 +79,39 @@
 	</appender>
 
 	<logger name="com.alibaba.otter.canal.instance" additivity="false">
-		<level value="INFO" />
+		<level value="DEBUG" />
 		<appender-ref ref="CANAL-ROOT" />
 	</logger>
 	<logger name="com.alibaba.otter.canal.deployer" additivity="false">
-		<level value="INFO" />
+		<level value="DEBUG" />
 		<appender-ref ref="CANAL-ROOT" />
 	</logger>
 	<logger name="com.alibaba.otter.canal.meta.FileMixedMetaManager" additivity="false">
-		<level value="INFO" />
+		<level value="DEBUG" />
 		<appender-ref ref="CANAL-META" />
 	</logger>
 	<logger name="com.alibaba.otter.canal.connector.kafka" additivity="false">
-		<level value="INFO" />
+		<level value="DEBUG" />
 		<appender-ref ref="CANAL-ROOT" />
 	</logger>
 	<logger name="com.alibaba.otter.canal.connector.rocketmq" additivity="false">
-		<level value="INFO" />
+		<level value="DEBUG" />
 		<appender-ref ref="CANAL-ROOT" />
 	</logger>
 	<logger name="com.alibaba.otter.canal.connector.rabbitmq" additivity="false">
-		<level value="INFO" />
+		<level value="DEBUG" />
 		<appender-ref ref="CANAL-ROOT" />
 	</logger>
 	<logger name="RocketmqClient" additivity="false">
-		<level value="INFO" />
+		<level value="DEBUG" />
 		<appender-ref ref="RocketmqClientAppender" />
   </logger>
 	<logger name="com.alibaba.otter.canal.connector.pulsarmq" additivity="false">
-		<level value="INFO" />
+		<level value="DEBUG" />
 		<appender-ref ref="CANAL-ROOT" />
 	</logger>
-	<root level="WARN">
-		<!-- <appender-ref ref="STDOUT"/>  -->
+	<root level="DEBUG">
+		<appender-ref ref="STDOUT"/>
 		<appender-ref ref="CANAL-ROOT" />
 	</root>
 </configuration>

smtweb-framework/canal/server/src/main/resources/spring/default-instance.xml

@@ -206,6 +206,5 @@
 		<property name="partitionsNum" value="${canal.mq.partitionsNum}" />
 		<property name="partitionHash" value="${canal.mq.partitionHash}" />
 		<property name="dynamicTopicPartitionNum" value="${canal.mq.dynamicTopicPartitionNum}" />
-		<property name="enableDynamicQueuePartition" value="${canal.mq.enableDynamicQueuePartition}" />
 	</bean>
 </beans>

smtweb-framework/canal/server/src/main/resources/spring/file-instance.xml

@@ -192,6 +192,5 @@
         <property name="partitionsNum" value="${canal.mq.partitionsNum}" />
         <property name="partitionHash" value="${canal.mq.partitionHash}" />
 		<property name="dynamicTopicPartitionNum" value="${canal.mq.dynamicTopicPartitionNum}" />
-		<property name="enableDynamicQueuePartition" value="${canal.mq.enableDynamicQueuePartition}" />
 	</bean>
 </beans>

smtweb-framework/canal/server/src/main/resources/spring/group-instance.xml

@@ -287,6 +287,5 @@
         <property name="partitionsNum" value="${canal.mq.partitionsNum}" />
         <property name="partitionHash" value="${canal.mq.partitionHash}" />
 		<property name="dynamicTopicPartitionNum" value="${canal.mq.dynamicTopicPartitionNum}" />
-		<property name="enableDynamicQueuePartition" value="${canal.mq.enableDynamicQueuePartition}" />
     </bean>
 </beans>

smtweb-framework/canal/server/src/main/resources/spring/memory-instance.xml

@@ -180,6 +180,5 @@
 		<property name="partitionsNum" value="${canal.mq.partitionsNum}" />
 		<property name="partitionHash" value="${canal.mq.partitionHash}" />
 		<property name="dynamicTopicPartitionNum" value="${canal.mq.dynamicTopicPartitionNum}" />
-		<property name="enableDynamicQueuePartition" value="${canal.mq.enableDynamicQueuePartition}" />
 	</bean>
 </beans>