> eventSink = null;
protected CanalEventFilter eventFilter = null;
protected CanalEventFilter eventBlackFilter = null;
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java
index 331b4d45..a0b32177 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java
@@ -2,6 +2,7 @@ package com.alibaba.otter.canal.parse.inbound.mysql;
import java.io.IOException;
+import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.parse.CanalEventParser;
@@ -9,7 +10,6 @@ import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache;
-import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
import com.alibaba.otter.canal.protocol.position.LogPosition;
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/AbstractLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/AbstractLogPositionManager.java
new file mode 100644
index 00000000..2c3c95da
--- /dev/null
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/AbstractLogPositionManager.java
@@ -0,0 +1,10 @@
+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 {
+}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/CanalLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/CanalLogPositionManager.java
index 25f63d15..7947231d 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/CanalLogPositionManager.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/CanalLogPositionManager.java
@@ -5,14 +5,13 @@ 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;
+
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/FailbackLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/FailbackLogPositionManager.java
index 6fa8efa5..fd2066f6 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/FailbackLogPositionManager.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/FailbackLogPositionManager.java
@@ -1,74 +1,81 @@
package com.alibaba.otter.canal.parse.index;
-import org.springframework.util.Assert;
-
-import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.protocol.position.LogPosition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
+ * Created by yinxiu on 17/3/18.
+ * Email: marklin.hz@gmail.com
+ *
* 实现基于failover查找的机制完成meta的操作
- *
+ *
*
* 应用场景：比如针对内存buffer，出现HA切换，先尝试从内存buffer区中找到lastest position，如果不存在才尝试找一下meta里消费的信息
*
- *
- * @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()) {
+ primary.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;
- }
-
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java
index e2c15401..072721af 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java
@@ -1,60 +1,82 @@
package com.alibaba.otter.canal.parse.index;
+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;
+import org.apache.commons.io.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.File;
import java.io.IOException;
import java.nio.charset.Charset;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
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.protocol.position.LogPosition;
-import com.google.common.base.Function;
-import com.google.common.collect.MigrateMap;
-
/**
+ * Created by yinxiu on 17/3/18.
+ * Email: marklin.hz@gmail.com
+ *
* 基于文件刷新的log position实现
- *
+ *
*
* 策略：
* 1. 先写内存，然后定时刷新数据到File
* 2. 数据采取overwrite模式(只保留最后一次)
*
- *
- * @author jianghang 2013-4-15 下午09:40:48
- * @version 1.0.4
*/
-public class FileMixedLogPositionManager extends MemoryLogPositionManager {
+public class FileMixedLogPositionManager extends AbstractLogPositionManager {
- 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 dataFileCaches;
- private ScheduledExecutorService executor;
- @SuppressWarnings("serial")
- private final LogPosition nullPosition = new LogPosition() {
- };
+ private final static Logger logger = LoggerFactory.getLogger(FileMixedLogPositionManager.class);
+ private final static Charset charset = Charset.forName("UTF-8");
- private long period = 1000; // 单位ms
- private Set persistTasks;
+ private File dataDir;
+ private Map dataFileCaches;
+
+ private ScheduledExecutorService executorService;
+
+ private final LogPosition nullPosition = new LogPosition(){};
+
+ private MemoryLogPositionManager memoryLogPositionManager;
+
+ private long period;
+ private Set 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() {
+
+ public File apply(String destination) {
+ return getDataFile(destination);
+ }
+ });
+
+ this.executorService = Executors.newScheduledThreadPool(1);
+ this.persistTasks = Collections.synchronizedSet(new HashSet());
+ }
+
+ @Override
public void start() {
super.start();
- Assert.notNull(dataDir);
if (!dataDir.exists()) {
try {
FileUtils.forceMkdir(dataDir);
@@ -67,30 +89,12 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager {
throw new CanalMetaManagerException("dir[" + dataDir.getPath() + "] can not read/write");
}
- dataFileCaches = MigrateMap.makeComputingMap(new Function() {
-
- public File apply(String destination) {
- return getDataFile(destination);
- }
- });
-
- executor = Executors.newScheduledThreadPool(1);
- positions = MigrateMap.makeComputingMap(new Function() {
-
- public LogPosition apply(String destination) {
- LogPosition logPosition = loadDataFromFile(dataFileCaches.get(destination));
- if (logPosition == null) {
- return nullPosition;
- } else {
- return logPosition;
- }
- }
- });
-
- persistTasks = Collections.synchronizedSet(new HashSet());
+ if (!memoryLogPositionManager.isStart()) {
+ memoryLogPositionManager.start();
+ }
// 启动定时工作任务
- executor.scheduleAtFixedRate(new Runnable() {
+ executorService.scheduleAtFixedRate(new Runnable() {
public void run() {
List tasks = new ArrayList(persistTasks);
@@ -106,28 +110,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 +153,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 +168,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 +191,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;
- }
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/MemoryLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/MemoryLogPositionManager.java
index c95fa4e2..9e20de5a 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/MemoryLogPositionManager.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/MemoryLogPositionManager.java
@@ -1,39 +1,44 @@
package com.alibaba.otter.canal.parse.index;
-import java.util.Map;
-
-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;
+import java.util.Map;
+import java.util.Set;
+
/**
- * 基于内存的实现
- *
- * @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 positions;
+ private Map 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 destinations() {
+ return positions.keySet();
+ }
+
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManager.java
index 3af97411..f3dd5479 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManager.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManager.java
@@ -1,55 +1,59 @@
package com.alibaba.otter.canal.parse.index;
-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;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.CollectionUtils;
+
+import java.util.List;
/**
- * 基于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 clientIdentitys = metaManager.listAllSubscribeInfo(destination);
+ List 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);
}
-
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java
index 0016bb06..1e36c6e7 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java
@@ -1,90 +1,83 @@
package com.alibaba.otter.canal.parse.index;
+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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.util.Assert;
-
-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 (!zooKeeperLogPositionManager.isStart()) {
- zooKeeperLogPositionManager.start();
- }
- executor = Executors.newFixedThreadPool(1);
- positions = MigrateMap.makeComputingMap(new Function() {
-
- public LogPosition apply(String destination) {
- LogPosition logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination);
- if (logPosition == null) {
- return nullPosition;
- } else {
- return logPosition;
- }
- }
- });
+ memoryLogPositionManager.start();
+ zooKeeperLogPositionManager.start();
}
+ @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);
- executor.submit(new Runnable() {
+ @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;
- }
-
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java
index e947720a..63a1dae3 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java
@@ -1,64 +1,81 @@
package com.alibaba.otter.canal.parse.index;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
+import com.alibaba.otter.canal.protocol.position.LogPosition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.util.Assert;
-
-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 static final Logger logger = LoggerFactory.getLogger(PeriodMixedLogPositionManager.class);
+
+ private MemoryLogPositionManager memoryLogPositionManager;
private ZooKeeperLogPositionManager zooKeeperLogPositionManager;
- private ScheduledExecutorService executor;
- @SuppressWarnings("serial")
- private final LogPosition nullPosition = new LogPosition() {
- };
+ private ScheduledExecutorService executorService;
- private long period = 1000; // 单位ms
- private Set persistTasks;
+ private long period;
+ private Set persistTasks;
+ private final LogPosition nullPosition = new LogPosition(){};
+
+ 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());
+ 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() {
-
- public LogPosition apply(String destination) {
- LogPosition logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination);
- if (logPosition == null) {
- return nullPosition;
- } else {
- return logPosition;
- }
- }
- });
-
- persistTasks = Collections.synchronizedSet(new HashSet());
// 启动定时工作任务
- executor.scheduleAtFixedRate(new Runnable() {
+ executorService.scheduleAtFixedRate(new Runnable() {
public void run() {
List tasks = new ArrayList(persistTasks);
@@ -76,36 +93,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;
- }
-
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/ZooKeeperLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/ZooKeeperLogPositionManager.java
index 996297ac..116a45e6 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/ZooKeeperLogPositionManager.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/ZooKeeperLogPositionManager.java
@@ -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;
+import org.I0Itec.zkclient.exception.ZkNoNodeException;
/**
- * 基于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;
- }
-
}
diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java
index b632639c..dc3de102 100644
--- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java
+++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java
@@ -2,13 +2,13 @@ package com.alibaba.otter.canal.parse.inbound.group;
import java.net.InetSocketAddress;
+import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import org.junit.Test;
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.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
@@ -70,15 +70,16 @@ 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;
}
diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogDumpTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogDumpTest.java
index 5ccd816c..ffa8d11f 100644
--- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogDumpTest.java
+++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogDumpTest.java
@@ -4,11 +4,12 @@ import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.List;
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
+import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import org.junit.Assert;
import org.junit.Test;
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;
@@ -80,17 +81,17 @@ public class LocalBinlogDumpTest {
}
});
- controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
+ controller.setLogPositionManager(new AbstractLogPositionManager() {
+ @Override
+ public LogPosition getLatestIndexBy(String destination) {
+ return null;
+ }
- public void persistLogPosition(String destination, LogPosition logPosition) {
- System.out.println(logPosition);
- }
-
- @Override
- public LogPosition getLatestIndexBy(String destination) {
- return null;
- }
- });
+ @Override
+ public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
+ System.out.println(logPosition);
+ }
+ });
controller.start();
diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java
index c2badf00..c17db374 100644
--- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java
+++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java
@@ -7,13 +7,13 @@ import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
+import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import com.alibaba.otter.canal.parse.helper.TimeoutChecker;
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);
diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java
index f4e54dfe..6c399e40 100644
--- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java
+++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java
@@ -4,11 +4,12 @@ import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.List;
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
+import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import org.junit.Assert;
import org.junit.Test;
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;
@@ -83,16 +84,16 @@ 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();
diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParserTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParserTest.java
index fbbc595c..3eaacd82 100644
--- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParserTest.java
+++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParserTest.java
@@ -5,12 +5,13 @@ import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
+import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import org.junit.Assert;
import org.junit.Test;
import com.alibaba.otter.canal.parse.helper.TimeoutChecker;
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,16 @@ 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 +137,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 +206,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 +282,7 @@ public class MysqlEventParserTest {
}
});
- controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
+ controller.setLogPositionManager(new AbstractLogPositionManager() {
public void persistLogPosition(String destination, LogPosition logPosition) {
System.out.println(logPosition);
diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManagerTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManagerTest.java
index ac1449b4..5d185648 100644
--- a/parse/src/test/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManagerTest.java
+++ b/parse/src/test/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManagerTest.java
@@ -26,17 +26,15 @@ 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);
diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/stub/AbstractCanalLogPositionManager.java b/parse/src/test/java/com/alibaba/otter/canal/parse/stub/AbstractCanalLogPositionManager.java
deleted file mode 100644
index df73735a..00000000
--- a/parse/src/test/java/com/alibaba/otter/canal/parse/stub/AbstractCanalLogPositionManager.java
+++ /dev/null
@@ -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 {
-
-}