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

Merge branch 'master' into master

Browse Source
This commit is contained in:
agapple
2017-07-04 21:49:22 +08:00
committed by GitHub
parent 9ba93bae5f ecdd619da7
commit 22e3d2f3f1
86 changed files with 22816 additions and 22429 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
parse/pom.xml
View File

@@ -3,7 +3,7 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.24-SNAPSHOT</version>
<version>1.0.25-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<artifactId>canal.parse</artifactId>

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

@@ -191,6 +191,9 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
} catch (TableIdNotFoundException e) {
throw e;
} catch (Throwable e) {
if (e.getCause() instanceof TableIdNotFoundException) {
throw (TableIdNotFoundException) e.getCause();
}
// 记录一下，出错的位点信息
processSinkError(e,
this.lastPosition,

41
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java
View File

@@ -83,30 +83,27 @@ public class LocalBinLogConnection implements ErosaConnection {
context.setLogPosition(new LogPosition(binlogfilename, binlogPosition));
while (running) {
boolean needContinue = true;
LogEvent event = null;
L:while (fetcher.fetch()) {
/*event = decoder.decode(fetcher, context);
if (event == null) {
throw new CanalParseException("parse failed");
}
LogEvent event = null;
L: while (fetcher.fetch()) {
/*
* event = decoder.decode(fetcher, context); if (event ==
* null) { throw new CanalParseException("parse failed"); }
* if (!func.sink(event)) { needContinue = false; break; }
*/
if (!func.sink(event)) {
needContinue = false;
break;
}*/
do {
if (event == null) {
event = new RotateLogEvent(context.getLogPosition().getFileName(), context.getLogPosition()
.getPosition());
} else {
event = decoder.decode(fetcher, context);
}
do {
if (event == null) {
event = new RotateLogEvent(context.getLogPosition().getFileName(), context.getLogPosition().getPosition());
} else {
event = decoder.decode(fetcher, context);
}
if (event != null && !func.sink(event)) {
needContinue = false;
break L;
}
} while (event != null);
if (event != null && !func.sink(event)) {
needContinue = false;
break L;
}
} while (event != null);
}
if (needContinue) {// 读取下一个

36
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java
View File

@@ -27,7 +27,7 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
private EntryPosition masterPosition; // binlog信息
private MysqlConnection metaConnection; // 查询meta信息的链接
private TableMetaCache tableMetaCache; // 对应meta
private String directory;
private boolean needWait = false;
private int bufferSize = 16 * 1024;
@@ -39,24 +39,24 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
@Override
protected ErosaConnection buildErosaConnection() {
return buildLocalBinLogConnection();
}
}
@Override
protected void preDump(ErosaConnection connection) {
metaConnection = buildMysqlConnection();
protected void preDump(ErosaConnection connection) {
metaConnection = buildMysqlConnection();
try {
metaConnection.connect();
} catch (IOException e) {
throw new CanalParseException(e);
}
tableMetaCache = new TableMetaCache(metaConnection);
((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache);
}
}
@Override
protected void afterDump(ErosaConnection connection) {
if (metaConnection != null) {
@Override
protected void afterDump(ErosaConnection connection) {
if (metaConnection != null) {
try {
metaConnection.disconnect();
} catch (IOException e) {
@@ -64,19 +64,19 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
.getAddress(), e);
}
}
}
}
public void start() throws CanalParseException {
public void start() throws CanalParseException {
if (runningInfo == null) { // 第一次链接主库
runningInfo = masterInfo;
}
super.start();
}
@Override
public void stop() {
if (metaConnection != null) {
public void stop() {
if (metaConnection != null) {
try {
metaConnection.disconnect();
} catch (IOException e) {
@@ -90,9 +90,9 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
}
super.stop();
}
}
private ErosaConnection buildLocalBinLogConnection() {
private ErosaConnection buildLocalBinLogConnection() {
LocalBinLogConnection connection = new LocalBinLogConnection();
connection.setBufferSize(this.bufferSize);
@@ -114,7 +114,7 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
connection.setCharset(connectionCharset);
return connection;
}
@Override
protected EntryPosition findStartPosition(ErosaConnection connection) {
// 处理逻辑

11
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
View File

@@ -139,7 +139,6 @@ public class MysqlConnection implements ErosaConnection {
binlogDumpHeader.setPacketBodyLength(cmdBody.length);
binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(),cmdBody);
connector.setDumping(true);
}
@@ -199,6 +198,16 @@ public class MysqlConnection implements ErosaConnection {
logger.warn(ExceptionUtils.getFullStackTrace(e));
}
try {
// 参考:https://github.com/alibaba/canal/issues/284
// mysql5.6需要设置slave_uuid避免被server kill链接
update("set @slave_uuid=uuid()");
} catch (Exception e) {
if (!StringUtils.contains(e.getMessage(), "Unknown system variable")) {
logger.warn(ExceptionUtils.getFullStackTrace(e));
}
}
try {
// mariadb针对特殊的类型，需要设置session变量
update("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'");

6
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java
View File

@@ -190,9 +190,9 @@ public class BinLogFileQueue {
files.addAll(FileUtils.listFiles(directory, new IOFileFilter() {
public boolean accept(File file) {
Pattern pattern = Pattern.compile("\\d+$");
Matcher matcher = pattern.matcher(file.getName());
return file.getName().startsWith(baseName) && matcher.find();
Pattern pattern = Pattern.compile("\\d+$");
Matcher matcher = pattern.matcher(file.getName());
return file.getName().startsWith(baseName) && matcher.find();
}
public boolean accept(File dir, String name) {

9
parse/src/main/java/com/alibaba/otter/canal/parse/index/AbstractLogPositionManager.java Normal file
View File

@@ -0,0 +1,9 @@
package com.alibaba.otter.canal.parse.index;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
/**
* Created by yinxiu on 17/3/17. Email: marklin.hz@gmail.com
*/
public abstract class AbstractLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager {
}

6
parse/src/main/java/com/alibaba/otter/canal/parse/index/CanalLogPositionManager.java
View File

@@ -5,14 +5,12 @@ import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.protocol.position.LogPosition;
/**
* 接口组合
*
* @author jianghang 2012-7-7 上午10:02:02
* @version 1.0.0
* Created by yinxiu on 17/3/17. Email: marklin.hz@gmail.com
*/
public interface CanalLogPositionManager extends CanalLifeCycle {
LogPosition getLatestIndexBy(String destination);
void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException;
}

67
parse/src/main/java/com/alibaba/otter/canal/parse/index/FailbackLogPositionManager.java
View File

@@ -1,74 +1,83 @@
package com.alibaba.otter.canal.parse.index;
import org.springframework.util.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.protocol.position.LogPosition;
/**
* Created by yinxiu on 17/3/18. Email: marklin.hz@gmail.com
* 实现基于failover查找的机制完成meta的操作
*
*
* <pre>
* 应用场景：比如针对内存buffer，出现HA切换，先尝试从内存buffer区中找到lastest position，如果不存在才尝试找一下meta里消费的信息
* </pre>
*
* @author jianghang 2012-7-20 下午02:33:20
*/
public class FailbackLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager {
public class FailbackLogPositionManager extends AbstractLogPositionManager {
private CanalLogPositionManager primary;
private CanalLogPositionManager failback;
private final static Logger logger = LoggerFactory.getLogger(FailbackLogPositionManager.class);
private final CanalLogPositionManager primary;
private final CanalLogPositionManager secondary;
public FailbackLogPositionManager(CanalLogPositionManager primary, CanalLogPositionManager secondary){
if (primary == null) {
throw new NullPointerException("nul primary LogPositionManager");
}
if (secondary == null) {
throw new NullPointerException("nul secondary LogPositionManager");
}
this.primary = primary;
this.secondary = secondary;
}
@Override
public void start() {
super.start();
Assert.notNull(primary);
Assert.notNull(failback);
if (!primary.isStart()) {
primary.start();
}
if (!failback.isStart()) {
failback.start();
if (!secondary.isStart()) {
secondary.start();
}
}
@Override
public void stop() {
super.stop();
if (secondary.isStart()) {
secondary.stop();
}
if (primary.isStart()) {
primary.stop();
}
if (failback.isStart()) {
failback.stop();
}
}
@Override
public LogPosition getLatestIndexBy(String destination) {
LogPosition logPosition = primary.getLatestIndexBy(destination);
if (logPosition == null) {
return failback.getLatestIndexBy(destination);
} else {
if (logPosition != null) {
return logPosition;
}
return secondary.getLatestIndexBy(destination);
}
@Override
public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
try {
primary.persistLogPosition(destination, logPosition);
} catch (CanalParseException e) {
failback.persistLogPosition(destination, logPosition);
logger.warn("persistLogPosition use primary log position manager exception. destination: {}, logPosition: {}",
destination,
logPosition,
e);
secondary.persistLogPosition(destination, logPosition);
}
}
public void setPrimary(CanalLogPositionManager primary) {
this.primary = primary;
}
public void setFailback(CanalLogPositionManager failback) {
this.failback = failback;
}
}

129
parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java
View File

@@ -16,45 +16,72 @@ import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import com.alibaba.otter.canal.common.utils.JsonUtils;
import com.alibaba.otter.canal.meta.exception.CanalMetaManagerException;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.protocol.position.LogPosition;
import com.google.common.base.Function;
import com.google.common.collect.MigrateMap;
/**
* 基于文件刷新的log position实现
*
* Created by yinxiu on 17/3/18. Email: marklin.hz@gmail.com 基于文件刷新的log
* position实现
*
* <pre>
* 策略：
* 1. 先写内存，然后定时刷新数据到File
* 2. 数据采取overwrite模式(只保留最后一次)
* </pre>
*
* @author jianghang 2013-4-15 下午09:40:48
* @version 1.0.4
*/
public class FileMixedLogPositionManager extends MemoryLogPositionManager {
public class FileMixedLogPositionManager extends AbstractLogPositionManager {
private final static Logger logger = LoggerFactory.getLogger(FileMixedLogPositionManager.class);
private final static Charset charset = Charset.forName("UTF-8");
private static final Logger logger = LoggerFactory.getLogger(FileMixedLogPositionManager.class);
private static final Charset charset = Charset.forName("UTF-8");
private File dataDir;
private String dataFileName = "parse.dat";
private Map<String, File> dataFileCaches;
private ScheduledExecutorService executor;
@SuppressWarnings("serial")
private ScheduledExecutorService executorService;
private final LogPosition nullPosition = new LogPosition() {
};
private long period = 1000; // 单位ms
private MemoryLogPositionManager memoryLogPositionManager;
private long period;
private Set<String> persistTasks;
public FileMixedLogPositionManager(File dataDir, long period, MemoryLogPositionManager memoryLogPositionManager){
if (dataDir == null) {
throw new NullPointerException("null dataDir");
}
if (period <= 0) {
throw new IllegalArgumentException("period must be positive, given: " + period);
}
if (memoryLogPositionManager == null) {
throw new NullPointerException("null memoryLogPositionManager");
}
this.dataDir = dataDir;
this.period = period;
this.memoryLogPositionManager = memoryLogPositionManager;
this.dataFileCaches = MigrateMap.makeComputingMap(new Function<String, File>() {
public File apply(String destination) {
return getDataFile(destination);
}
});
this.executorService = Executors.newScheduledThreadPool(1);
this.persistTasks = Collections.synchronizedSet(new HashSet<String>());
}
@Override
public void start() {
super.start();
Assert.notNull(dataDir);
if (!dataDir.exists()) {
try {
FileUtils.forceMkdir(dataDir);
@@ -67,30 +94,12 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager {
throw new CanalMetaManagerException("dir[" + dataDir.getPath() + "] can not read/write");
}
dataFileCaches = MigrateMap.makeComputingMap(new Function<String, File>() {
public File apply(String destination) {
return getDataFile(destination);
}
});
executor = Executors.newScheduledThreadPool(1);
positions = MigrateMap.makeComputingMap(new Function<String, LogPosition>() {
public LogPosition apply(String destination) {
LogPosition logPosition = loadDataFromFile(dataFileCaches.get(destination));
if (logPosition == null) {
return nullPosition;
} else {
return logPosition;
}
}
});
persistTasks = Collections.synchronizedSet(new HashSet<String>());
if (!memoryLogPositionManager.isStart()) {
memoryLogPositionManager.start();
}
// 启动定时工作任务
executor.scheduleAtFixedRate(new Runnable() {
executorService.scheduleAtFixedRate(new Runnable() {
public void run() {
List<String> tasks = new ArrayList<String>(persistTasks);
@@ -106,28 +115,35 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager {
}
}
}, period, period, TimeUnit.MILLISECONDS);
}
@Override
public void stop() {
super.stop();
flushDataToFile();
executor.shutdownNow();
positions.clear();
}
public void persistLogPosition(String destination, LogPosition logPosition) {
persistTasks.add(destination);// 添加到任务队列中进行触发
super.persistLogPosition(destination, logPosition);
executorService.shutdown();
memoryLogPositionManager.stop();
}
@Override
public LogPosition getLatestIndexBy(String destination) {
LogPosition logPostion = super.getLatestIndexBy(destination);
if (logPostion == nullPosition) {
return null;
} else {
return logPostion;
LogPosition logPosition = memoryLogPositionManager.getLatestIndexBy(destination);
if (logPosition != null) {
return logPosition;
}
logPosition = loadDataFromFile(dataFileCaches.get(destination));
if (logPosition == null) {
return nullPosition;
}
return logPosition;
}
@Override
public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
persistTasks.add(destination);
memoryLogPositionManager.persistLogPosition(destination, logPosition);
}
// ============================ helper method ======================
@@ -142,11 +158,12 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager {
}
}
String dataFileName = "parse.dat";
return new File(destinationMetaDir, dataFileName);
}
private void flushDataToFile() {
for (String destination : positions.keySet()) {
for (String destination : memoryLogPositionManager.destinations()) {
flushDataToFile(destination);
}
}
@@ -156,7 +173,7 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager {
}
private void flushDataToFile(String destination, File dataFile) {
LogPosition position = positions.get(destination);
LogPosition position = memoryLogPositionManager.getLatestIndexBy(destination);
if (position != null && position != nullPosition) {
String json = JsonUtils.marshalToString(position);
try {
@@ -179,16 +196,4 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager {
throw new CanalMetaManagerException(e);
}
}
public void setDataDir(String dataDir) {
this.dataDir = new File(dataDir);
}
public void setDataDir(File dataDir) {
this.dataDir = dataDir;
}
public void setPeriod(long period) {
this.period = period;
}
}

24
parse/src/main/java/com/alibaba/otter/canal/parse/index/MemoryLogPositionManager.java
View File

@@ -1,39 +1,43 @@
package com.alibaba.otter.canal.parse.index;
import java.util.Map;
import java.util.Set;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.protocol.position.LogPosition;
import com.google.common.collect.MapMaker;
/**
* 基于内存的实现
*
* @author jianghang 2012-7-7 上午10:17:23
* @version 1.0.0
* Created by yinxiu on 17/3/17. Email: marklin.hz@gmail.com
*/
public class MemoryLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager {
public class MemoryLogPositionManager extends AbstractLogPositionManager {
protected Map<String, LogPosition> positions;
private Map<String, LogPosition> positions;
@Override
public void start() {
super.start();
positions = new MapMaker().makeMap();
}
@Override
public void stop() {
super.stop();
positions.clear();
}
@Override
public LogPosition getLatestIndexBy(String destination) {
return positions.get(destination);
}
public void persistLogPosition(String destination, LogPosition logPosition) {
@Override
public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
positions.put(destination, logPosition);
}
public Set<String> destinations() {
return positions.keySet();
}
}

51
parse/src/main/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManager.java
View File

@@ -4,52 +4,56 @@ import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.meta.CanalMetaManager;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.position.LogPosition;
import com.alibaba.otter.canal.store.helper.CanalEventUtils;
/**
* 基于meta信息的实现
*
* @author jianghang 2012-7-10 下午05:02:33
* @version 1.0.0
* Created by yinxiu on 17/3/18. Email: marklin.hz@gmail.com
*/
public class MetaLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager {
public class MetaLogPositionManager extends AbstractLogPositionManager {
private static final Logger logger = LoggerFactory.getLogger(MetaLogPositionManager.class);
private CanalMetaManager metaManager;
private final static Logger logger = LoggerFactory.getLogger(MetaLogPositionManager.class);
public void start() {
super.start();
Assert.notNull(metaManager);
if (!metaManager.isStart()) {
metaManager.start();
private final CanalMetaManager metaManager;
public MetaLogPositionManager(CanalMetaManager metaManager){
if (metaManager == null) {
throw new NullPointerException("null metaManager");
}
this.metaManager = metaManager;
}
@Override
public void stop() {
super.stop();
if (metaManager.isStart()) {
metaManager.stop();
}
}
public void persistLogPosition(String destination, LogPosition logPosition) {
// do nothing
logger.info("persist LogPosition:{}", destination, logPosition);
@Override
public void start() {
super.start();
if (!metaManager.isStart()) {
metaManager.start();
}
}
@Override
public LogPosition getLatestIndexBy(String destination) {
List<ClientIdentity> clientIdentitys = metaManager.listAllSubscribeInfo(destination);
List<ClientIdentity> clientIdentities = metaManager.listAllSubscribeInfo(destination);
LogPosition result = null;
if (!CollectionUtils.isEmpty(clientIdentitys)) {
if (!CollectionUtils.isEmpty(clientIdentities)) {
// 尝试找到一个最小的logPosition
for (ClientIdentity clientIdentity : clientIdentitys) {
for (ClientIdentity clientIdentity : clientIdentities) {
LogPosition position = (LogPosition) metaManager.getCursor(clientIdentity);
if (position == null) {
continue;
@@ -66,8 +70,9 @@ public class MetaLogPositionManager extends AbstractCanalLifeCycle implements Ca
return result;
}
public void setMetaManager(CanalMetaManager metaManager) {
this.metaManager = metaManager;
@Override
public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
// do nothing
logger.info("persist LogPosition:{}", destination, logPosition);
}
}

101
parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java
View File

@@ -5,86 +5,83 @@ import java.util.concurrent.Executors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.protocol.position.LogPosition;
import com.google.common.base.Function;
import com.google.common.collect.MigrateMap;
/**
* 混合memory + zookeeper的存储模式
*
* @author jianghang 2012-7-7 上午10:33:19
* @version 1.0.0
* Created by yinxiu on 17/3/17. Email: marklin.hz@gmail.com Memory first.
* Asynchronous commit position info to ZK.
*/
public class MixedLogPositionManager extends MemoryLogPositionManager implements CanalLogPositionManager {
public class MixedLogPositionManager extends AbstractLogPositionManager {
private static final Logger logger = LoggerFactory.getLogger(MixedLogPositionManager.class);
private ZooKeeperLogPositionManager zooKeeperLogPositionManager;
private ExecutorService executor;
@SuppressWarnings("serial")
private final LogPosition nullPosition = new LogPosition() {
};
private final Logger logger = LoggerFactory.getLogger(MixedLogPositionManager.class);
private final MemoryLogPositionManager memoryLogPositionManager;
private final ZooKeeperLogPositionManager zooKeeperLogPositionManager;
private final ExecutorService executor;
public MixedLogPositionManager(ZkClientx zkClient){
if (zkClient == null) {
throw new NullPointerException("null zkClient");
}
this.memoryLogPositionManager = new MemoryLogPositionManager();
this.zooKeeperLogPositionManager = new ZooKeeperLogPositionManager(zkClient);
this.executor = Executors.newFixedThreadPool(1);
}
@Override
public void start() {
super.start();
Assert.notNull(zooKeeperLogPositionManager);
if (!memoryLogPositionManager.isStart()) {
memoryLogPositionManager.start();
}
if (!zooKeeperLogPositionManager.isStart()) {
zooKeeperLogPositionManager.start();
}
executor = Executors.newFixedThreadPool(1);
positions = MigrateMap.makeComputingMap(new Function<String, LogPosition>() {
public LogPosition apply(String destination) {
LogPosition logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination);
if (logPosition == null) {
return nullPosition;
} else {
return logPosition;
}
}
});
}
@Override
public void stop() {
super.stop();
if (zooKeeperLogPositionManager.isStart()) {
zooKeeperLogPositionManager.stop();
}
executor.shutdownNow();
positions.clear();
executor.shutdown();
zooKeeperLogPositionManager.stop();
memoryLogPositionManager.stop();
}
public void persistLogPosition(final String destination, final LogPosition logPosition) {
super.persistLogPosition(destination, logPosition);
@Override
public LogPosition getLatestIndexBy(String destination) {
LogPosition logPosition = memoryLogPositionManager.getLatestIndexBy(destination);
if (logPosition != null) {
return logPosition;
}
logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination);
// 这里保持和重构前的逻辑一致,重新添加到Memory中
if (logPosition != null) {
memoryLogPositionManager.persistLogPosition(destination, logPosition);
}
return logPosition;
}
@Override
public void persistLogPosition(final String destination, final LogPosition logPosition) throws CanalParseException {
memoryLogPositionManager.persistLogPosition(destination, logPosition);
executor.submit(new Runnable() {
public void run() {
try {
zooKeeperLogPositionManager.persistLogPosition(destination, logPosition);
} catch (Exception e) {
logger.error("ERROR # persist to zookeepr has an error", e);
logger.error("ERROR # persist to zookeeper has an error", e);
}
}
});
}
public LogPosition getLatestIndexBy(String destination) {
LogPosition logPosition = super.getLatestIndexBy(destination);
if (logPosition == nullPosition) {
return null;
} else {
return logPosition;
}
}
// ======================== setter / getter ======================
public void setZooKeeperLogPositionManager(ZooKeeperLogPositionManager zooKeeperLogPositionManager) {
this.zooKeeperLogPositionManager = zooKeeperLogPositionManager;
}
}

112
parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java
View File

@@ -11,54 +11,77 @@ import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.protocol.position.LogPosition;
import com.google.common.base.Function;
import com.google.common.collect.MigrateMap;
/**
* 基于定时刷新的策略的mixed实现
*
* @author jianghang 2012-9-12 上午11:18:14
* @version 1.0.0
* Created by yinxiu on 17/3/18. Email: marklin.hz@gmail.com
*/
public class PeriodMixedLogPositionManager extends MemoryLogPositionManager implements CanalLogPositionManager {
public class PeriodMixedLogPositionManager extends AbstractLogPositionManager {
private static final Logger logger = LoggerFactory.getLogger(PeriodMixedLogPositionManager.class);
private MemoryLogPositionManager memoryLogPositionManager;
private ZooKeeperLogPositionManager zooKeeperLogPositionManager;
private ScheduledExecutorService executor;
@SuppressWarnings("serial")
private ScheduledExecutorService executorService;
private long period;
private Set<String> persistTasks;
private final LogPosition nullPosition = new LogPosition() {
};
private long period = 1000; // 单位ms
private Set<String> persistTasks;
public PeriodMixedLogPositionManager(MemoryLogPositionManager memoryLogPositionManager,
ZooKeeperLogPositionManager zooKeeperLogPositionManager, long period){
if (memoryLogPositionManager == null) {
throw new NullPointerException("null memoryLogPositionManager");
}
if (zooKeeperLogPositionManager == null) {
throw new NullPointerException("null zooKeeperLogPositionManager");
}
if (period <= 0) {
throw new IllegalArgumentException("period must be positive, given: " + period);
}
this.memoryLogPositionManager = memoryLogPositionManager;
this.zooKeeperLogPositionManager = zooKeeperLogPositionManager;
this.period = period;
this.persistTasks = Collections.synchronizedSet(new HashSet<String>());
this.executorService = Executors.newScheduledThreadPool(1);
}
@Override
public void stop() {
super.stop();
if (zooKeeperLogPositionManager.isStart()) {
zooKeeperLogPositionManager.stop();
}
if (memoryLogPositionManager.isStart()) {
memoryLogPositionManager.stop();
}
executorService.shutdown();
}
@Override
public void start() {
super.start();
Assert.notNull(zooKeeperLogPositionManager);
if (!memoryLogPositionManager.isStart()) {
memoryLogPositionManager.start();
}
if (!zooKeeperLogPositionManager.isStart()) {
zooKeeperLogPositionManager.start();
}
executor = Executors.newScheduledThreadPool(1);
positions = MigrateMap.makeComputingMap(new Function<String, LogPosition>() {
public LogPosition apply(String destination) {
LogPosition logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination);
if (logPosition == null) {
return nullPosition;
} else {
return logPosition;
}
}
});
persistTasks = Collections.synchronizedSet(new HashSet<String>());
// 启动定时工作任务
executor.scheduleAtFixedRate(new Runnable() {
executorService.scheduleAtFixedRate(new Runnable() {
public void run() {
List<String> tasks = new ArrayList<String>(persistTasks);
@@ -76,36 +99,19 @@ public class PeriodMixedLogPositionManager extends MemoryLogPositionManager impl
}, period, period, TimeUnit.MILLISECONDS);
}
public void stop() {
super.stop();
if (zooKeeperLogPositionManager.isStart()) {
zooKeeperLogPositionManager.stop();
}
executor.shutdownNow();
positions.clear();
}
public void persistLogPosition(String destination, LogPosition logPosition) {
persistTasks.add(destination);// 添加到任务队列中进行触发
super.persistLogPosition(destination, logPosition);
}
@Override
public LogPosition getLatestIndexBy(String destination) {
LogPosition logPostion = super.getLatestIndexBy(destination);
if (logPostion == nullPosition) {
LogPosition logPosition = memoryLogPositionManager.getLatestIndexBy(destination);
if (logPosition == nullPosition) {
return null;
} else {
return logPostion;
return logPosition;
}
}
public void setZooKeeperLogPositionManager(ZooKeeperLogPositionManager zooKeeperLogPositionManager) {
this.zooKeeperLogPositionManager = zooKeeperLogPositionManager;
@Override
public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
persistTasks.add(destination);
memoryLogPositionManager.persistLogPosition(destination, logPosition);
}
public void setPeriod(long period) {
this.period = period;
}
}

35
parse/src/main/java/com/alibaba/otter/canal/parse/index/ZooKeeperLogPositionManager.java
View File

@@ -1,33 +1,29 @@
package com.alibaba.otter.canal.parse.index;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.springframework.util.Assert;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.common.utils.JsonUtils;
import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.protocol.position.LogPosition;
/**
* 基于zk的实现
*
* @author jianghang 2012-7-7 上午10:08:27
* @version 1.0.0
* Created by yinxiu on 17/3/17. Email: marklin.hz@gmail.com
*/
public class ZooKeeperLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager {
public class ZooKeeperLogPositionManager extends AbstractLogPositionManager {
private ZkClientx zkClientx;
private final ZkClientx zkClientx;
public void start() {
super.start();
Assert.notNull(zkClientx);
}
public void stop() {
super.stop();
public ZooKeeperLogPositionManager(ZkClientx zkClient){
if (zkClient == null) {
throw new NullPointerException("null zkClient");
}
this.zkClientx = zkClient;
}
@Override
public LogPosition getLatestIndexBy(String destination) {
String path = ZookeeperPathUtils.getParsePath(destination);
byte[] data = zkClientx.readData(path, true);
@@ -38,7 +34,8 @@ public class ZooKeeperLogPositionManager extends AbstractCanalLifeCycle implemen
return JsonUtils.unmarshalFromByte(data, LogPosition.class);
}
public void persistLogPosition(String destination, LogPosition logPosition) {
@Override
public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
String path = ZookeeperPathUtils.getParsePath(destination);
byte[] data = JsonUtils.marshalToByte(logPosition);
try {
@@ -48,10 +45,4 @@ public class ZooKeeperLogPositionManager extends AbstractCanalLifeCycle implemen
}
}
// ================== setter / getter =================
public void setZkClientx(ZkClientx zkClientx) {
this.zkClientx = zkClientx;
}
}

1
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/EventTransactionBufferTest.java
View File

@@ -6,7 +6,6 @@ import java.util.Date;
import java.util.List;
import org.junit.Assert;
import org.junit.Test;
import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer.TransactionFlushCallback;

14
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java
View File

@@ -8,7 +8,7 @@ import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.inbound.AbstractBinlogParser;
import com.alibaba.otter.canal.parse.inbound.BinlogParser;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager;
import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
@@ -70,15 +70,17 @@ public class GroupEventPaserTest {
mysqlEventPaser.setMasterPosition(defaultPosition);
mysqlEventPaser.setBinlogParser(buildParser(buildAuthentication()));
mysqlEventPaser.setEventSink(new EntryEventSink());
mysqlEventPaser.setLogPositionManager(new AbstractCanalLogPositionManager() {
public void persistLogPosition(String destination, LogPosition logPosition) {
// System.out.println(logPosition);
}
mysqlEventPaser.setLogPositionManager(new AbstractLogPositionManager() {
@Override
public LogPosition getLatestIndexBy(String destination) {
return null;
}
@Override
public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
System.out.println(logPosition);
}
});
return mysqlEventPaser;
}

15
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogDumpTest.java
View File

@@ -5,11 +5,11 @@ import java.nio.charset.Charset;
import java.util.List;
import org.junit.Assert;
import org.junit.Test;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest;
import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
@@ -81,16 +81,17 @@ public class LocalBinlogDumpTest {
}
});
controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
public void persistLogPosition(String destination, LogPosition logPosition) {
System.out.println(logPosition);
}
controller.setLogPositionManager(new AbstractLogPositionManager() {
@Override
public LogPosition getLatestIndexBy(String destination) {
return null;
}
@Override
public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
System.out.println(logPosition);
}
});
controller.start();

8
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java
View File

@@ -12,8 +12,8 @@ import org.junit.Before;
import org.junit.Test;
import com.alibaba.otter.canal.parse.helper.TimeoutChecker;
import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest;
import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
@@ -70,7 +70,7 @@ public class LocalBinlogEventParserTest {
});
controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
controller.setLogPositionManager(new AbstractLogPositionManager() {
public void persistLogPosition(String destination, LogPosition logPosition) {
System.out.println(logPosition);
@@ -133,7 +133,7 @@ public class LocalBinlogEventParserTest {
}
});
controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
controller.setLogPositionManager(new AbstractLogPositionManager() {
public void persistLogPosition(String destination, LogPosition logPosition) {
System.out.println(logPosition);
@@ -198,7 +198,7 @@ public class LocalBinlogEventParserTest {
}
});
controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
controller.setLogPositionManager(new AbstractLogPositionManager() {
public void persistLogPosition(String destination, LogPosition logPosition) {
System.out.println(logPosition);

15
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java
View File

@@ -5,11 +5,11 @@ import java.nio.charset.Charset;
import java.util.List;
import org.junit.Assert;
import org.junit.Test;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest;
import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
@@ -84,16 +84,17 @@ public class MysqlDumpTest {
}
});
controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
public void persistLogPosition(String destination, LogPosition logPosition) {
System.out.println(logPosition);
}
controller.setLogPositionManager(new AbstractLogPositionManager() {
@Override
public LogPosition getLatestIndexBy(String destination) {
return null;
}
@Override
public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
System.out.println(logPosition);
}
});
controller.start();

20
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParserTest.java
View File

@@ -8,9 +8,10 @@ import java.util.concurrent.atomic.AtomicLong;
import org.junit.Assert;
import org.junit.Test;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.helper.TimeoutChecker;
import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest;
import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
@@ -67,16 +68,17 @@ public class MysqlEventParserTest {
}
});
controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
public void persistLogPosition(String destination, LogPosition logPosition) {
System.out.println(logPosition);
}
controller.setLogPositionManager(new AbstractLogPositionManager() {
@Override
public LogPosition getLatestIndexBy(String destination) {
return null;
}
@Override
public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
System.out.println(logPosition);
}
});
controller.start();
@@ -136,7 +138,7 @@ public class MysqlEventParserTest {
}
});
controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
controller.setLogPositionManager(new AbstractLogPositionManager() {
public void persistLogPosition(String destination, LogPosition logPosition) {
System.out.println(logPosition);
@@ -205,7 +207,7 @@ public class MysqlEventParserTest {
}
});
controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
controller.setLogPositionManager(new AbstractLogPositionManager() {
public void persistLogPosition(String destination, LogPosition logPosition) {
System.out.println(logPosition);
@@ -281,7 +283,7 @@ public class MysqlEventParserTest {
}
});
controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
controller.setLogPositionManager(new AbstractLogPositionManager() {
public void persistLogPosition(String destination, LogPosition logPosition) {
System.out.println(logPosition);

2
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/SimpleDdlParserTest.java
View File

@@ -164,7 +164,7 @@ public class SimpleDdlParserTest {
Assert.assertEquals("retl_mark1", result.getOriTableName());
Assert.assertEquals("retl_mark3", result.getTableName());
//正则匹配test case
// 正则匹配test case
queryString = "rename table totl_mark to totl_mark2";
result = SimpleDdlParser.parse(queryString, "retl");

17
parse/src/test/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManagerTest.java
View File

@@ -3,9 +3,8 @@ package com.alibaba.otter.canal.parse.index;
import java.io.File;
import java.io.IOException;
import org.junit.Assert;
import org.apache.commons.io.FileUtils;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -27,17 +26,19 @@ public class FileMixedLogPositionManagerTest extends AbstractLogPositionManagerT
@Test
public void testAll() {
FileMixedLogPositionManager logPositionManager = new FileMixedLogPositionManager();
logPositionManager.setDataDir(dataDir);
logPositionManager.setPeriod(100);
MemoryLogPositionManager memoryLogPositionManager = new MemoryLogPositionManager();
FileMixedLogPositionManager logPositionManager = new FileMixedLogPositionManager(dataDir,
1000,
memoryLogPositionManager);
logPositionManager.start();
LogPosition position2 = doTest(logPositionManager);
sleep(1500);
FileMixedLogPositionManager logPositionManager2 = new FileMixedLogPositionManager();
logPositionManager2.setDataDir(dataDir);
logPositionManager2.setPeriod(100);
FileMixedLogPositionManager logPositionManager2 = new FileMixedLogPositionManager(dataDir,
1000,
memoryLogPositionManager);
logPositionManager2.start();
LogPosition getPosition2 = logPositionManager2.getLatestIndexBy(destination);

6
parse/src/test/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManagerTest.java
View File

@@ -3,9 +3,8 @@ package com.alibaba.otter.canal.parse.index;
import java.net.InetSocketAddress;
import java.util.Date;
import org.junit.Assert;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -46,8 +45,7 @@ public class MetaLogPositionManagerTest extends AbstractLogPositionManagerTest {
metaManager.setZooKeeperMetaManager(zooKeeperMetaManager);
metaManager.start();
MetaLogPositionManager logPositionManager = new MetaLogPositionManager();
logPositionManager.setMetaManager(metaManager);
MetaLogPositionManager logPositionManager = new MetaLogPositionManager(metaManager);
logPositionManager.start();
// 构建meta信息
ClientIdentity client1 = new ClientIdentity(destination, (short) 1);

14
parse/src/test/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManagerTest.java
View File

@@ -1,8 +1,7 @@
package com.alibaba.otter.canal.parse.index;
import org.junit.Assert;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -28,19 +27,16 @@ public class MixedLogPositionManagerTest extends AbstractLogPositionManagerTest
@Test
public void testAll() {
MixedLogPositionManager logPositionManager = new MixedLogPositionManager();
MemoryLogPositionManager memoryLogPositionManager = new MemoryLogPositionManager();
ZooKeeperLogPositionManager zookeeperLogPositionManager = new ZooKeeperLogPositionManager(zkclientx);
ZooKeeperLogPositionManager zookeeperLogPositionManager = new ZooKeeperLogPositionManager();
zookeeperLogPositionManager.setZkClientx(zkclientx);
logPositionManager.setZooKeeperLogPositionManager(zookeeperLogPositionManager);
MixedLogPositionManager logPositionManager = new MixedLogPositionManager(zkclientx);
logPositionManager.start();
LogPosition position2 = doTest(logPositionManager);
sleep(1000);
MixedLogPositionManager logPositionManager2 = new MixedLogPositionManager();
logPositionManager2.setZooKeeperLogPositionManager(zookeeperLogPositionManager);
MixedLogPositionManager logPositionManager2 = new MixedLogPositionManager(zkclientx);
logPositionManager2.start();
LogPosition getPosition2 = logPositionManager2.getLatestIndexBy(destination);

13
parse/src/test/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManagerTest.java
View File

@@ -1,8 +1,7 @@
package com.alibaba.otter.canal.parse.index;
import org.junit.Assert;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -28,19 +27,17 @@ public class PeriodMixedLogPositionManagerTest extends AbstractLogPositionManage
@Test
public void testAll() {
PeriodMixedLogPositionManager logPositionManager = new PeriodMixedLogPositionManager();
MemoryLogPositionManager memoryLogPositionManager = new MemoryLogPositionManager();
ZooKeeperLogPositionManager zookeeperLogPositionManager = new ZooKeeperLogPositionManager(zkclientx);
ZooKeeperLogPositionManager zookeeperLogPositionManager = new ZooKeeperLogPositionManager();
zookeeperLogPositionManager.setZkClientx(zkclientx);
PeriodMixedLogPositionManager logPositionManager = new PeriodMixedLogPositionManager(memoryLogPositionManager, zookeeperLogPositionManager, 1000L);
logPositionManager.setZooKeeperLogPositionManager(zookeeperLogPositionManager);
logPositionManager.start();
LogPosition position2 = doTest(logPositionManager);
sleep(1500);
PeriodMixedLogPositionManager logPositionManager2 = new PeriodMixedLogPositionManager();
logPositionManager2.setZooKeeperLogPositionManager(zookeeperLogPositionManager);
PeriodMixedLogPositionManager logPositionManager2 = new PeriodMixedLogPositionManager(memoryLogPositionManager, zookeeperLogPositionManager, 1000L);
logPositionManager2.start();
LogPosition getPosition2 = logPositionManager2.getLatestIndexBy(destination);

3
parse/src/test/java/com/alibaba/otter/canal/parse/index/ZooKeeperLogPositionManagerTest.java
View File

@@ -25,8 +25,7 @@ public class ZooKeeperLogPositionManagerTest extends AbstractLogPositionManagerT
@Test
public void testAll() {
ZooKeeperLogPositionManager logPositionManager = new ZooKeeperLogPositionManager();
logPositionManager.setZkClientx(zkclientx);
ZooKeeperLogPositionManager logPositionManager = new ZooKeeperLogPositionManager(zkclientx);
logPositionManager.start();
doTest(logPositionManager);

8
parse/src/test/java/com/alibaba/otter/canal/parse/stub/AbstractCanalLogPositionManager.java
View File

@@ -1,8 +0,0 @@
package com.alibaba.otter.canal.parse.stub;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
public abstract class AbstractCanalLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager {
}