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

remove duplicate code

Browse Source
This commit is contained in:
yinxiu
2016-01-14 10:59:40 +08:00
parent ed74d934d3
commit d2f98cda87
2 changed files with 19 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -120,7 +120,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
throw new CanalParseException("consume failed!");
}
LogPosition position = buildLastTranasctionPosition(transaction);
LogPosition position = buildLastTransactionPosition(transaction);
if (position != null) { // 可能position为空
logPositionManager.persistLogPosition(AbstractEventParser.this.destination, position);
}
@@ -336,20 +336,11 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
return profilingEnabled.get();
}
protected LogPosition buildLastTranasctionPosition(List<CanalEntry.Entry> entries) { // 初始化一下
protected LogPosition buildLastTransactionPosition(List<CanalEntry.Entry> entries) { // 初始化一下
for (int i = entries.size() - 1; i > 0; i--) {
CanalEntry.Entry entry = entries.get(i);
if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {// 尽量记录一个事务做为position
LogPosition logPosition = new LogPosition();
EntryPosition position = new EntryPosition();
position.setJournalName(entry.getHeader().getLogfileName());
position.setPosition(entry.getHeader().getLogfileOffset());
position.setTimestamp(entry.getHeader().getExecuteTime());
logPosition.setPostion(position);
LogIdentity identity = new LogIdentity(runningInfo.getAddress(), -1L);
logPosition.setIdentity(identity);
return logPosition;
return buildLastPosition(entry);
}
}

32
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java
View File

@@ -700,22 +700,22 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
}
}
protected Entry parseAndProfilingIfNecessary(LogEvent bod) throws Exception {
long startTs = -1;
boolean enabled = getProfilingEnabled();
if (enabled) {
startTs = System.currentTimeMillis();
}
CanalEntry.Entry event = binlogParser.parse(bod);
if (enabled) {
this.parsingInterval = System.currentTimeMillis() - startTs;
}
if (parsedEventCount.incrementAndGet() < 0) {
parsedEventCount.set(0);
}
return event;
}
// protected Entry parseAndProfilingIfNecessary(LogEvent bod) throws Exception {
// long startTs = -1;
// boolean enabled = getProfilingEnabled();
// if (enabled) {
// startTs = System.currentTimeMillis();
// }
// CanalEntry.Entry event = binlogParser.parse(bod);
// if (enabled) {
// this.parsingInterval = System.currentTimeMillis() - startTs;
// }
//
// if (parsedEventCount.incrementAndGet() < 0) {
// parsedEventCount.set(0);
// }
// return event;
// }
public void setSupportBinlogFormats(String formatStrs) {
String[] formats = StringUtils.split(formatStrs, ',');