chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Parallel参数细节的完善

Browse Source
This commit is contained in:
Chuanyi Li
2018-07-05 20:24:42 +08:00
parent da8bf59d5b
commit be605617c3
7 changed files with 16 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
deployer/src/main/resources/canal.properties
View File

@@ -53,7 +53,9 @@ canal.instance.get.ddl.isolation = false
# parallel parser config # parallel parser config
canal.instance.parser.parallel = true 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 canal.instance.parser.parallelBufferSize = 256
################################################# #################################################

2
deployer/src/main/resources/spring/default-instance.xml
View File

@@ -195,7 +195,7 @@
<!-- parallel parser --> <!-- parallel parser -->
<property name="parallel" value="${canal.instance.parser.parallel:true}" /> <property name="parallel" value="${canal.instance.parser.parallel:true}" />
<property name="parallelThreadSize" value="${canal.instance.parser.parallelThreadSize:16}" /> <property name="parallelThreadSize" value="${canal.instance.parser.parallelThreadSize}" />
<property name="parallelBufferSize" value="${canal.instance.parser.parallelBufferSize:256}" /> <property name="parallelBufferSize" value="${canal.instance.parser.parallelBufferSize:256}" />
</bean> </bean>
</beans> </beans>

2
deployer/src/main/resources/spring/file-instance.xml
View File

@@ -180,7 +180,7 @@
<!-- parallel parser --> <!-- parallel parser -->
<property name="parallel" value="${canal.instance.parser.parallel:true}" /> <property name="parallel" value="${canal.instance.parser.parallel:true}" />
<property name="parallelThreadSize" value="${canal.instance.parser.parallelThreadSize:16}" /> <property name="parallelThreadSize" value="${canal.instance.parser.parallelThreadSize}" />
<property name="parallelBufferSize" value="${canal.instance.parser.parallelBufferSize:256}" /> <property name="parallelBufferSize" value="${canal.instance.parser.parallelBufferSize:256}" />
</bean> </bean>
</beans> </beans>

4
deployer/src/main/resources/spring/group-instance.xml
View File

@@ -169,7 +169,7 @@
<!-- parallel parser --> <!-- parallel parser -->
<property name="parallel" value="${canal.instance.parser.parallel:true}" /> <property name="parallel" value="${canal.instance.parser.parallel:true}" />
<property name="parallelThreadSize" value="${canal.instance.parser.parallelThreadSize:16}" /> <property name="parallelThreadSize" value="${canal.instance.parser.parallelThreadSize}" />
<property name="parallelBufferSize" value="${canal.instance.parser.parallelBufferSize:256}" /> <property name="parallelBufferSize" value="${canal.instance.parser.parallelBufferSize:256}" />
</bean> </bean>
@@ -268,7 +268,7 @@
<!-- parallel parser --> <!-- parallel parser -->
<property name="parallel" value="${canal.instance.parser.parallel:true}" /> <property name="parallel" value="${canal.instance.parser.parallel:true}" />
<property name="parallelThreadSize" value="${canal.instance.parser.parallelThreadSize:16}" /> <property name="parallelThreadSize" value="${canal.instance.parser.parallelThreadSize}" />
<property name="parallelBufferSize" value="${canal.instance.parser.parallelBufferSize:256}" /> <property name="parallelBufferSize" value="${canal.instance.parser.parallelBufferSize:256}" />
</bean> </bean>
</beans> </beans>

2
deployer/src/main/resources/spring/local-instance.xml
View File

@@ -140,7 +140,7 @@
<!-- parallel parser --> <!-- parallel parser -->
<property name="parallel" value="${canal.instance.parser.parallel:true}" /> <property name="parallel" value="${canal.instance.parser.parallel:true}" />
<property name="parallelThreadSize" value="${canal.instance.parser.parallelThreadSize:16}" /> <property name="parallelThreadSize" value="${canal.instance.parser.parallelThreadSize}" />
<property name="parallelBufferSize" value="${canal.instance.parser.parallelBufferSize:256}" /> <property name="parallelBufferSize" value="${canal.instance.parser.parallelBufferSize:256}" />
</bean> </bean>
</beans> </beans>

2
deployer/src/main/resources/spring/memory-instance.xml
View File

@@ -168,7 +168,7 @@
<!-- parallel parser --> <!-- parallel parser -->
<property name="parallel" value="${canal.instance.parser.parallel:true}" /> <property name="parallel" value="${canal.instance.parser.parallel:true}" />
<property name="parallelThreadSize" value="${canal.instance.parser.parallelThreadSize:16}" /> <property name="parallelThreadSize" value="${canal.instance.parser.parallelThreadSize}" />
<property name="parallelBufferSize" value="${canal.instance.parser.parallelBufferSize:256}" /> <property name="parallelBufferSize" value="${canal.instance.parser.parallelBufferSize:256}" />
</bean> </bean>
</beans> </beans>

12
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
View File

@@ -90,9 +90,9 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
protected boolean isGTIDMode = false; // 是否是GTID模式 protected boolean isGTIDMode = false; // 是否是GTID模式
protected boolean parallel = true; // 是否开启并行解析模式 protected boolean parallel = true; // 是否开启并行解析模式
protected int parallelThreadSize = Runtime.getRuntime() protected Integer parallelThreadSize = Runtime.getRuntime()
.availableProcessors() * 60 / 100; // 60%的能力跑解析,剩余部分处理网络 .availableProcessors() * 60 / 100; // 60%的能力跑解析,剩余部分处理网络
protected int parallelBufferSize = 16 * parallelThreadSize; protected int parallelBufferSize = 256; // 必须为2的幂
protected MultiStageCoprocessor multiStageCoprocessor; protected MultiStageCoprocessor multiStageCoprocessor;
protected abstract BinlogParser buildParser(); protected abstract BinlogParser buildParser();
@@ -595,11 +595,13 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
return parallelThreadSize; return parallelThreadSize;
} }
public void setParallelThreadSize(int parallelThreadSize) { public void setParallelThreadSize(Integer parallelThreadSize) {
this.parallelThreadSize = parallelThreadSize; if (parallelThreadSize != null) {
this.parallelThreadSize = parallelThreadSize;
}
} }
public int getParallelBufferSize() { public Integer getParallelBufferSize() {
return parallelBufferSize; return parallelBufferSize;
} }