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

skip tsdb snapshot apply if has not receive new ddl change from last snapshot time

Browse Source
This commit is contained in:
wuwo
2018-11-06 16:58:50 +08:00
parent 8ee546d34d
commit f2b8d000f7
1 changed files with 4 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java
View File

@@ -63,6 +63,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
private CanalEventFilter filter; private CanalEventFilter filter;
private CanalEventFilter blackFilter; private CanalEventFilter blackFilter;
private EntryPosition lastPosition; private EntryPosition lastPosition;
private boolean hasNewDdl;
private MetaHistoryDAO metaHistoryDAO; private MetaHistoryDAO metaHistoryDAO;
private MetaSnapshotDAO metaSnapshotDAO; private MetaSnapshotDAO metaSnapshotDAO;
private int snapshotInterval = 24; private int snapshotInterval = 24;
@@ -139,6 +140,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
synchronized (memoryTableMeta) { synchronized (memoryTableMeta) {
if (memoryTableMeta.apply(position, schema, ddl, extra)) { if (memoryTableMeta.apply(position, schema, ddl, extra)) {
this.lastPosition = position; this.lastPosition = position;
this.hasNewDdl = true;
// 同步每次变更给远程做历史记录 // 同步每次变更给远程做历史记录
return applyHistoryToDB(position, schema, ddl, extra); return applyHistoryToDB(position, schema, ddl, extra);
} else { } else {
@@ -270,10 +272,11 @@ public class DatabaseTableMeta implements TableMetaTSDB {
MemoryTableMeta tmpMemoryTableMeta = new MemoryTableMeta(); MemoryTableMeta tmpMemoryTableMeta = new MemoryTableMeta();
Map<String, String> schemaDdls = null; Map<String, String> schemaDdls = null;
synchronized (memoryTableMeta) { synchronized (memoryTableMeta) {
if (!init && position == null) { if (!init && !hasNewDdl) {
// 如果是持续构建,则识别一下是否有DDL变更过,如果没有就忽略了 // 如果是持续构建,则识别一下是否有DDL变更过,如果没有就忽略了
return false; return false;
} }
this.hasNewDdl = false;
schemaDdls = memoryTableMeta.snapshot(); schemaDdls = memoryTableMeta.snapshot();
for (Map.Entry<String, String> entry : schemaDdls.entrySet()) { for (Map.Entry<String, String> entry : schemaDdls.entrySet()) {
tmpMemoryTableMeta.apply(position, entry.getKey(), entry.getValue(), null); tmpMemoryTableMeta.apply(position, entry.getKey(), entry.getValue(), null);