optimizer filter

This commit is contained in:
agapple
2018-12-06 10:50:12 +08:00
parent 1b11aa1d21
commit c045dc244e
5 changed files with 34 additions and 14 deletions

View File

@@ -125,4 +125,8 @@ public class AviaterRegexFilter implements CanalEventFilter<String> {
return result; return result;
} }
@Override
public String toString() {
return pattern;
}
} }

View File

@@ -11,6 +11,7 @@ import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
import com.alibaba.otter.canal.parse.inbound.BinlogParser; import com.alibaba.otter.canal.parse.inbound.BinlogParser;
import com.alibaba.otter.canal.parse.inbound.MultiStageCoprocessor; import com.alibaba.otter.canal.parse.inbound.MultiStageCoprocessor;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert; import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DatabaseTableMeta;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DefaultTableMetaTSDBFactory; import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DefaultTableMetaTSDBFactory;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB; import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDBFactory; import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDBFactory;
@@ -64,9 +65,30 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
super.setEventFilter(eventFilter); super.setEventFilter(eventFilter);
// 触发一下filter变更 // 触发一下filter变更
if (eventFilter != null && eventFilter instanceof AviaterRegexFilter && binlogParser instanceof LogEventConvert) { if (eventFilter != null && eventFilter instanceof AviaterRegexFilter) {
if (binlogParser instanceof LogEventConvert) {
((LogEventConvert) binlogParser).setNameFilter((AviaterRegexFilter) eventFilter); ((LogEventConvert) binlogParser).setNameFilter((AviaterRegexFilter) eventFilter);
} }
if (tableMetaTSDB != null && tableMetaTSDB instanceof DatabaseTableMeta) {
((DatabaseTableMeta) tableMetaTSDB).setFilter(eventFilter);
}
}
}
public void setEventBlackFilter(CanalEventFilter eventBlackFilter) {
super.setEventBlackFilter(eventBlackFilter);
// 触发一下filter变更
if (eventBlackFilter != null && eventBlackFilter instanceof AviaterRegexFilter) {
if (binlogParser instanceof LogEventConvert) {
((LogEventConvert) binlogParser).setNameBlackFilter((AviaterRegexFilter) eventBlackFilter);
}
if (tableMetaTSDB != null && tableMetaTSDB instanceof DatabaseTableMeta) {
((DatabaseTableMeta) tableMetaTSDB).setBlackFilter(eventBlackFilter);
}
}
} }
/** /**
@@ -115,16 +137,6 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
super.stop(); super.stop();
} }
public void setEventBlackFilter(CanalEventFilter eventBlackFilter) {
super.setEventBlackFilter(eventBlackFilter);
// 触发一下filter变更
if (eventBlackFilter != null && eventBlackFilter instanceof AviaterRegexFilter
&& binlogParser instanceof LogEventConvert) {
((LogEventConvert) binlogParser).setNameBlackFilter((AviaterRegexFilter) eventBlackFilter);
}
}
protected MultiStageCoprocessor buildMultiStageCoprocessor() { protected MultiStageCoprocessor buildMultiStageCoprocessor() {
MysqlMultiStageCoprocessor mysqlMultiStageCoprocessor = new MysqlMultiStageCoprocessor(parallelBufferSize, MysqlMultiStageCoprocessor mysqlMultiStageCoprocessor = new MysqlMultiStageCoprocessor(parallelBufferSize,
parallelThreadSize, parallelThreadSize,

View File

@@ -506,8 +506,8 @@ public class MysqlConnection implements ErosaConnection {
ResultSetPacket rs = null; ResultSetPacket rs = null;
try { try {
rs = query("select @@global.binlog_checksum"); rs = query("select @@global.binlog_checksum");
} catch (IOException e) { } catch (Throwable e) {
throw new CanalParseException(e); // ignore
} }
List<String> columnValues = rs.getFieldValues(); List<String> columnValues = rs.getFieldValues();

View File

@@ -974,10 +974,12 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
public void setNameFilter(AviaterRegexFilter nameFilter) { public void setNameFilter(AviaterRegexFilter nameFilter) {
this.nameFilter = nameFilter; this.nameFilter = nameFilter;
logger.warn("--> init table filter : " + nameFilter.toString());
} }
public void setNameBlackFilter(AviaterRegexFilter nameBlackFilter) { public void setNameBlackFilter(AviaterRegexFilter nameBlackFilter) {
this.nameBlackFilter = nameBlackFilter; this.nameBlackFilter = nameBlackFilter;
logger.warn("--> init table black filter : " + nameBlackFilter.toString());
} }
public void setTableMetaCache(TableMetaCache tableMetaCache) { public void setTableMetaCache(TableMetaCache tableMetaCache) {

View File

@@ -381,6 +381,7 @@
</configuration> </configuration>
</plugin> </plugin>
<!-- javadoc --> <!-- javadoc -->
<!--
<plugin> <plugin>
<groupId>org.apache.maven.plugins</groupId> <groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-javadoc-plugin</artifactId> <artifactId>maven-javadoc-plugin</artifactId>
@@ -399,6 +400,7 @@
<additionalparam>-Xdoclint:none</additionalparam> <additionalparam>-Xdoclint:none</additionalparam>
</configuration> </configuration>
</plugin> </plugin>
-->
</plugins> </plugins>
<sourceDirectory>src/main/java</sourceDirectory> <sourceDirectory>src/main/java</sourceDirectory>
<testSourceDirectory>src/test/java</testSourceDirectory> <testSourceDirectory>src/test/java</testSourceDirectory>