diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/position/PositionRange.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/position/PositionRange.java index abfac4b2..b2e9c83c 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/position/PositionRange.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/position/PositionRange.java @@ -19,6 +19,8 @@ public class PositionRange implements Serializable { // add by ljh at 2012-09-05，用于记录一个可被ack的位置，保证每次提交到cursor中的位置是一个完整事务的结束 private T ack; private T end; + // add by ljh at 2019-06-25，用于精确记录ringbuffer中的位点 + private Long endSeq = -1L; public PositionRange(){ } @@ -52,6 +54,14 @@ public class PositionRange implements Serializable { this.ack = ack; } + public Long getEndSeq() { + return endSeq; + } + + public void setEndSeq(Long endSeq) { + this.endSeq = endSeq; + } + @Override public String toString() { return ToStringBuilder.reflectionToString(this, CanalToStringStyle.DEFAULT_STYLE); diff --git a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java index b07062ee..7cf53c39 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java @@ -432,8 +432,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C } // 可定时清理数据 - canalInstance.getEventStore().ack(positionRanges.getEnd()); - + canalInstance.getEventStore().ack(positionRanges.getEnd(), positionRanges.getEndSeq()); } /** diff --git a/store/src/main/java/com/alibaba/otter/canal/store/CanalEventStore.java b/store/src/main/java/com/alibaba/otter/canal/store/CanalEventStore.java index d83bcba0..8a3fc714 100644 --- a/store/src/main/java/com/alibaba/otter/canal/store/CanalEventStore.java +++ b/store/src/main/java/com/alibaba/otter/canal/store/CanalEventStore.java @@ -76,6 +76,13 @@ public interface CanalEventStore extends CanalLifeCycle, CanalStoreScavenge { */ void ack(Position position) throws CanalStoreException; + /** + * 删除指定seqId之前的数据 + * + * @Since 1.1.4 + */ + void ack(Position position, Long seqId) throws CanalStoreException; + /** * 出错时执行回滚操作(未提交ack的所有状态信息重新归位，减少出错时数据全部重来的成本) */ diff --git a/store/src/main/java/com/alibaba/otter/canal/store/helper/CanalEventUtils.java b/store/src/main/java/com/alibaba/otter/canal/store/helper/CanalEventUtils.java index 4fdbfcba..7642c92a 100644 --- a/store/src/main/java/com/alibaba/otter/canal/store/helper/CanalEventUtils.java +++ b/store/src/main/java/com/alibaba/otter/canal/store/helper/CanalEventUtils.java @@ -70,6 +70,10 @@ public class CanalEventUtils { position.setPosition(event.getPosition()); position.setTimestamp(event.getExecuteTime()); position.setIncluded(included); + // add serverId at 2016-06-28 + position.setServerId(event.getServerId()); + // add gtid + position.setGtid(event.getGtid()); LogPosition logPosition = new LogPosition(); logPosition.setPostion(position); diff --git a/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java b/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java index 7f5eeb47..22308c71 100644 --- a/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java +++ b/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java @@ -334,6 +334,7 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple range.setStart(CanalEventUtils.createPosition(entrys.get(0))); range.setEnd(CanalEventUtils.createPosition(entrys.get(result.getEvents().size() - 1))); + range.setEndSeq(end); // 记录一下是否存在可以被ack的点 for (int i = entrys.size() - 1; i >= 0; i--) { @@ -369,9 +370,9 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple return CanalEventUtils.createPosition(event, false); } else if (firstSeqeuence > INIT_SEQUENCE && firstSeqeuence < putSequence.get()) { // ack未追上put操作 - Event event = entries[getIndex(firstSeqeuence + 1)]; // 最后一次ack的位置数据 - // + 1 - return CanalEventUtils.createPosition(event, true); + Event event = entries[getIndex(firstSeqeuence)]; // 最后一次ack的位置数据,需要移动到下一条,included + // = false + return CanalEventUtils.createPosition(event, false); } else if (firstSeqeuence > INIT_SEQUENCE && firstSeqeuence == putSequence.get()) { // 已经追上，store中没有数据 Event event = entries[getIndex(firstSeqeuence)]; // 最后一次ack的位置数据，和last为同一条，included @@ -410,10 +411,19 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple } public void ack(Position position) throws CanalStoreException { - cleanUntil(position); + cleanUntil(position, -1L); } + public void ack(Position position, Long seqId) throws CanalStoreException { + cleanUntil(position, seqId); + } + + @Override public void cleanUntil(Position position) throws CanalStoreException { + cleanUntil(position, -1L); + } + + public void cleanUntil(Position position, Long seqId) throws CanalStoreException { final ReentrantLock lock = this.lock; lock.lock(); try { @@ -425,6 +435,9 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple // ack没有list，但有已存在的foreach，还是节省一下list的开销 long localExecTime = 0L; int deltaRows = 0; + if (seqId > 0) { + maxSequence = seqId; + } for (long next = sequence + 1; next <= maxSequence; next++) { Event event = entries[getIndex(next)]; if (localExecTime == 0 && event.getExecuteTime() > 0) { @@ -432,8 +445,8 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple } deltaRows += event.getRowsCount(); memsize += calculateSize(event); - boolean match = CanalEventUtils.checkPosition(event, (LogPosition) position); - if (match) {// 找到对应的position，更新ack seq + if ((seqId < 0 || next == seqId) && CanalEventUtils.checkPosition(event, (LogPosition) position)) { + // 找到对应的position，更新ack seq hasMatch = true; if (batchMode.isMemSize()) { @@ -442,6 +455,12 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple for (long index = sequence + 1; index < next; index++) { entries[getIndex(index)] = null;// 设置为null } + + // 考虑getFirstPosition/getLastPosition会获取最后一次ack的position信息 + // ack清理的时候只处理entry=null，释放内存 + Event lastEvent = entries[getIndex(next)]; + lastEvent.setEntry(null); + lastEvent.setRawEntry(null); } if (ackSequence.compareAndSet(sequence, next)) {// 避免并发ack @@ -681,5 +700,4 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple return ackTableRows; } - } diff --git a/store/src/test/java/com/alibaba/otter/cancel/store/memory/buffer/MemoryEventStoreMemBatchTest.java b/store/src/test/java/com/alibaba/otter/cancel/store/memory/buffer/MemoryEventStoreMemBatchTest.java index 8e999502..77603fd3 100644 --- a/store/src/test/java/com/alibaba/otter/cancel/store/memory/buffer/MemoryEventStoreMemBatchTest.java +++ b/store/src/test/java/com/alibaba/otter/cancel/store/memory/buffer/MemoryEventStoreMemBatchTest.java @@ -300,10 +300,11 @@ public class MemoryEventStoreMemBatchTest extends MemoryEventStoreBase { first = eventStore.getFirstPosition(); lastest = eventStore.getLatestPosition(); List entrys = new ArrayList(entrys2.getEvents()); - Assert.assertEquals(first, entrys2.getPositionRange().getStart()); + // Assert.assertEquals(first, entrys2.getPositionRange().getStart()); Assert.assertEquals(lastest, entrys2.getPositionRange().getEnd()); - Assert.assertEquals(first, CanalEventUtils.createPosition(entrys.get(0))); + // Assert.assertEquals(first, + // CanalEventUtils.createPosition(entrys.get(0))); Assert.assertEquals(lastest, CanalEventUtils.createPosition(entrys.get(entrys.size() - 1))); // 全部ack掉