forked from alibaba/canal
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request alibaba#291 from MarkLinHz/master
重构LogPositionManager并适配XML配置
- Loading branch information
Showing
25 changed files
with
404 additions
and
410 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
10 changes: 10 additions & 0 deletions
10
parse/src/main/java/com/alibaba/otter/canal/parse/index/AbstractLogPositionManager.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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: [email protected] | ||
| */ | ||
| public abstract class AbstractLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager { | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,14 +5,13 @@ | |
| 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: [email protected] | ||
| */ | ||
| public interface CanalLogPositionManager extends CanalLifeCycle { | ||
|
|
||
| LogPosition getLatestIndexBy(String destination); | ||
|
|
||
| void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException; | ||
|
|
||
| } | ||
67 changes: 37 additions & 30 deletions
67
parse/src/main/java/com/alibaba/otter/canal/parse/index/FailbackLogPositionManager.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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: [email protected] | ||
| * | ||
| * 实现基于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 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"); | ||
| } | ||
|
|
||
| private CanalLogPositionManager primary; | ||
| private CanalLogPositionManager failback; | ||
| 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 (primary.isStart()) { | ||
| primary.stop(); | ||
| if (secondary.isStart()) { | ||
| secondary.stop(); | ||
| } | ||
|
|
||
| if (failback.isStart()) { | ||
| failback.stop(); | ||
| if (primary.isStart()) { | ||
| primary.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; | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.