diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties
index 90050148..23ec03a1 100644
--- a/deployer/src/main/resources/canal.properties
+++ b/deployer/src/main/resources/canal.properties
@@ -53,7 +53,9 @@ canal.instance.get.ddl.isolation = false
# parallel parser config
canal.instance.parser.parallel = true
-canal.instance.parser.parallelThreadSize = 16
+## 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
#################################################
diff --git a/deployer/src/main/resources/spring/default-instance.xml b/deployer/src/main/resources/spring/default-instance.xml
index 31299e66..c4465df1 100644
--- a/deployer/src/main/resources/spring/default-instance.xml
+++ b/deployer/src/main/resources/spring/default-instance.xml
@@ -195,7 +195,7 @@
-
+
diff --git a/deployer/src/main/resources/spring/file-instance.xml b/deployer/src/main/resources/spring/file-instance.xml
index 494f6c87..340a8158 100644
--- a/deployer/src/main/resources/spring/file-instance.xml
+++ b/deployer/src/main/resources/spring/file-instance.xml
@@ -180,7 +180,7 @@
-
+
diff --git a/deployer/src/main/resources/spring/group-instance.xml b/deployer/src/main/resources/spring/group-instance.xml
index f1add032..60d069dc 100644
--- a/deployer/src/main/resources/spring/group-instance.xml
+++ b/deployer/src/main/resources/spring/group-instance.xml
@@ -169,7 +169,7 @@
-
+
@@ -268,7 +268,7 @@
-
+
diff --git a/deployer/src/main/resources/spring/local-instance.xml b/deployer/src/main/resources/spring/local-instance.xml
index 3148572d..0b2f96aa 100644
--- a/deployer/src/main/resources/spring/local-instance.xml
+++ b/deployer/src/main/resources/spring/local-instance.xml
@@ -140,7 +140,7 @@
-
+
\ No newline at end of file
diff --git a/deployer/src/main/resources/spring/memory-instance.xml b/deployer/src/main/resources/spring/memory-instance.xml
index 7404229c..8fec8918 100644
--- a/deployer/src/main/resources/spring/memory-instance.xml
+++ b/deployer/src/main/resources/spring/memory-instance.xml
@@ -168,7 +168,7 @@
-
+
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
index 39da0163..54bb14e7 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
@@ -90,9 +90,9 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle
protected boolean isGTIDMode = false; // 是否是GTID模式
protected boolean parallel = true; // 是否开启并行解析模式
- protected int parallelThreadSize = Runtime.getRuntime()
+ protected Integer parallelThreadSize = Runtime.getRuntime()
.availableProcessors() * 60 / 100; // 60%的能力跑解析,剩余部分处理网络
- protected int parallelBufferSize = 16 * parallelThreadSize;
+ protected int parallelBufferSize = 256; // 必须为2的幂
protected MultiStageCoprocessor multiStageCoprocessor;
protected abstract BinlogParser buildParser();
@@ -595,11 +595,13 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle
return parallelThreadSize;
}
- public void setParallelThreadSize(int parallelThreadSize) {
- this.parallelThreadSize = parallelThreadSize;
+ public void setParallelThreadSize(Integer parallelThreadSize) {
+ if (parallelThreadSize != null) {
+ this.parallelThreadSize = parallelThreadSize;
+ }
}
- public int getParallelBufferSize() {
+ public Integer getParallelBufferSize() {
return parallelBufferSize;
}