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

init

Browse Source
This commit is contained in:
agapple
2014-09-23 15:11:35 +08:00
commit a0defe575b
331 changed files with 53031 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
store/pom.xml Normal file
View File

@@ -0,0 +1,36 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.0.19-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.store</artifactId>
<packaging>jar</packaging>
<name>canal store module for otter ${project.version}</name>
<dependencies>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.protocol</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.meta</artifactId>
<version>${project.version}</version>
</dependency>
<!-- test dependency -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

29
store/src/main/java/com/alibaba/otter/canal/store/AbstractCanalGroupStore.java Normal file
View File

@@ -0,0 +1,29 @@
package com.alibaba.otter.canal.store;
import java.util.Map;
import org.springframework.util.Assert;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.google.common.collect.MapMaker;
/**
* @author zebin.xuzb 2012-10-30 下午3:45:17
* @since 1.0.0
*/
public abstract class AbstractCanalGroupStore<T> extends AbstractCanalLifeCycle implements CanalGroupEventStore<T> {
protected Map<String, StoreInfo> stores = new MapMaker().makeMap();
@Override
public void addStoreInfo(StoreInfo info) {
checkInfo(info);
stores.put(info.getStoreName(), info);
}
protected void checkInfo(StoreInfo info) {
Assert.notNull(info);
Assert.hasText(info.getStoreName());
}
}

116
store/src/main/java/com/alibaba/otter/canal/store/AbstractCanalStoreScavenge.java Normal file
View File

@@ -0,0 +1,116 @@
package com.alibaba.otter.canal.store;
import java.util.List;
import org.springframework.util.CollectionUtils;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.meta.CanalMetaManager;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.position.LogPosition;
import com.alibaba.otter.canal.protocol.position.Position;
/**
* store回收机制
*
* @author jianghang 2012-8-8 下午12:57:36
* @version 1.0.0
*/
public abstract class AbstractCanalStoreScavenge extends AbstractCanalLifeCycle implements CanalStoreScavenge {
protected String destination;
protected CanalMetaManager canalMetaManager;
protected boolean onAck = true;
protected boolean onFull = false;
protected boolean onSchedule = false;
protected String scavengeSchedule = null;
public void scavenge() {
Position position = getLatestAckPosition(destination);
cleanUntil(position);
}
/**
* 找出该destination中可被清理掉的position位置
*
* @param destination
*/
private Position getLatestAckPosition(String destination) {
List<ClientIdentity> clientIdentitys = canalMetaManager.listAllSubscribeInfo(destination);
LogPosition result = null;
if (!CollectionUtils.isEmpty(clientIdentitys)) {
// 尝试找到一个最小的logPosition
for (ClientIdentity clientIdentity : clientIdentitys) {
LogPosition position = (LogPosition) canalMetaManager.getCursor(clientIdentity);
if (position == null) {
continue;
}
if (result == null) {
result = position;
} else {
result = min(result, position);
}
}
}
return result;
}
/**
* 找出一个最小的position位置
*/
private LogPosition min(LogPosition position1, LogPosition position2) {
if (position1.getIdentity().equals(position2.getIdentity())) {
// 首先根据文件进行比较
if (position1.getPostion().getJournalName().compareTo(position2.getPostion().getJournalName()) < 0) {
return position2;
} else if (position1.getPostion().getJournalName().compareTo(position2.getPostion().getJournalName()) > 0) {
return position1;
} else {
// 根据offest进行比较
if (position1.getPostion().getPosition() < position2.getPostion().getPosition()) {
return position2;
} else {
return position1;
}
}
} else {
// 不同的主备库，根据时间进行比较
if (position1.getPostion().getTimestamp() < position2.getPostion().getTimestamp()) {
return position2;
} else {
return position1;
}
}
}
public void setOnAck(boolean onAck) {
this.onAck = onAck;
}
public void setOnFull(boolean onFull) {
this.onFull = onFull;
}
public void setOnSchedule(boolean onSchedule) {
this.onSchedule = onSchedule;
}
public String getScavengeSchedule() {
return scavengeSchedule;
}
public void setScavengeSchedule(String scavengeSchedule) {
this.scavengeSchedule = scavengeSchedule;
}
public void setDestination(String destination) {
this.destination = destination;
}
public void setCanalMetaManager(CanalMetaManager canalMetaManager) {
this.canalMetaManager = canalMetaManager;
}
}

84
store/src/main/java/com/alibaba/otter/canal/store/CanalEventStore.java Normal file
View File

@@ -0,0 +1,84 @@
package com.alibaba.otter.canal.store;
import java.util.List;
import java.util.concurrent.TimeUnit;
import com.alibaba.otter.canal.common.CanalLifeCycle;
import com.alibaba.otter.canal.protocol.position.Position;
import com.alibaba.otter.canal.store.model.Events;
/**
* canel数据存储接口
*
* @author jianghang 2012-6-14 下午08:44:52
* @version 1.0.0
*/
public interface CanalEventStore<T> extends CanalLifeCycle, CanalStoreScavenge {
/**
* 添加一组数据对象，阻塞等待其操作完成 (比如一次性添加一个事务数据)
*/
void put(List<T> data) throws InterruptedException, CanalStoreException;
/**
* 添加一组数据对象，阻塞等待其操作完成或者时间超时 (比如一次性添加一个事务数据)
*/
boolean put(List<T> data, long timeout, TimeUnit unit) throws InterruptedException, CanalStoreException;
/**
* 添加一组数据对象 (比如一次性添加一个事务数据)
*/
boolean tryPut(List<T> data) throws CanalStoreException;
/**
* 添加一个数据对象，阻塞等待其操作完成
*/
void put(T data) throws InterruptedException, CanalStoreException;
/**
* 添加一个数据对象，阻塞等待其操作完成或者时间超时
*/
boolean put(T data, long timeout, TimeUnit unit) throws InterruptedException, CanalStoreException;
/**
* 添加一个数据对象
*/
boolean tryPut(T data) throws CanalStoreException;
/**
* 获取指定大小的数据，阻塞等待其操作完成
*/
Events<T> get(Position start, int batchSize) throws InterruptedException, CanalStoreException;
/**
* 获取指定大小的数据，阻塞等待其操作完成或者时间超时
*/
Events<T> get(Position start, int batchSize, long timeout, TimeUnit unit) throws InterruptedException,
CanalStoreException;
/**
* 根据指定位置，获取一个指定大小的数据
*/
Events<T> tryGet(Position start, int batchSize) throws CanalStoreException;
/**
* 获取最后一条数据的position
*/
Position getLatestPosition() throws CanalStoreException;
/**
* 获取第一条数据的position，如果没有数据返回为null
*/
Position getFirstPosition() throws CanalStoreException;
/**
* 删除{@linkplain Position}之前的数据
*/
void ack(Position position) throws CanalStoreException;
/**
* 出错时执行回滚操作(未提交ack的所有状态信息重新归位，减少出错时数据全部重来的成本)
*/
void rollback() throws CanalStoreException;
}

12
store/src/main/java/com/alibaba/otter/canal/store/CanalGroupEventStore.java Normal file
View File

@@ -0,0 +1,12 @@
package com.alibaba.otter.canal.store;
/**
* 提供给上层统一的 store 视图，内部则支持多种store混合，并且维持着多个store供上层进行路由
*
* @author zebin.xuzb 2012-10-30 下午12:17:26
* @since 1.0.0
*/
public interface CanalGroupEventStore<T> extends CanalEventStore<T> {
void addStoreInfo(StoreInfo info);
}

21
store/src/main/java/com/alibaba/otter/canal/store/CanalStoreConstants.java Normal file
View File

@@ -0,0 +1,21 @@
package com.alibaba.otter.canal.store;
/**
* 常量值
*
* @author jianghang 2012-6-14 下午09:40:33
* @version 1.0.0
*/
public interface CanalStoreConstants {
public static final String CODE_POSITION_NOT_FOUND = "position:%s not found";
public static final String CODE_POSITION_NOT_IN_ORDER = "position:%s not in order";
public static final String ENCODING = "utf8";
public static final int MAX_STORECOUNT = 100;
public static final int ROLLOVERCOUNT = 100;
}

35
store/src/main/java/com/alibaba/otter/canal/store/CanalStoreException.java Normal file
View File

@@ -0,0 +1,35 @@
package com.alibaba.otter.canal.store;
import com.alibaba.otter.canal.common.CanalException;
/**
* canal 异常定义
*
* @author jianghang 2012-6-15 下午04:57:35
* @version 1.0.0
*/
public class CanalStoreException extends CanalException {
private static final long serialVersionUID = -7288830284122672209L;
public CanalStoreException(String errorCode){
super(errorCode);
}
public CanalStoreException(String errorCode, Throwable cause){
super(errorCode, cause);
}
public CanalStoreException(String errorCode, String errorDesc){
super(errorCode + ":" + errorDesc);
}
public CanalStoreException(String errorCode, String errorDesc, Throwable cause){
super(errorCode + ":" + errorDesc, cause);
}
public CanalStoreException(Throwable cause){
super(cause);
}
}

22
store/src/main/java/com/alibaba/otter/canal/store/CanalStoreScavenge.java Normal file
View File

@@ -0,0 +1,22 @@
package com.alibaba.otter.canal.store;
import com.alibaba.otter.canal.protocol.position.Position;
/**
* store空间回收机制，信息采集以及控制何时调用{@linkplain CanalEventStore}.cleanUtil()接口
*
* @author jianghang 2012-8-8 上午11:57:42
* @version 1.0.0
*/
public interface CanalStoreScavenge {
/**
* 清理position之前的数据
*/
void cleanUntil(Position position) throws CanalStoreException;
/**
* 删除所有的数据
*/
void cleanAll() throws CanalStoreException;
}

28
store/src/main/java/com/alibaba/otter/canal/store/StoreInfo.java Normal file
View File

@@ -0,0 +1,28 @@
package com.alibaba.otter.canal.store;
/**
* @author zebin.xuzb 2012-10-30 下午1:05:13
* @since 1.0.0
*/
public class StoreInfo {
private String storeName;
private String filter;
public String getStoreName() {
return storeName;
}
public String getFilter() {
return filter;
}
public void setStoreName(String storeName) {
this.storeName = storeName;
}
public void setFilter(String filter) {
this.filter = filter;
}
}

93
store/src/main/java/com/alibaba/otter/canal/store/helper/CanalEventUtils.java Normal file
View File

@@ -0,0 +1,93 @@
package com.alibaba.otter.canal.store.helper;
import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
import com.alibaba.otter.canal.protocol.position.LogPosition;
import com.alibaba.otter.canal.store.model.Event;
/**
* 相关的操作工具
*
* @author jianghang 2012-6-19 下午05:49:21
* @version 1.0.0
*/
public class CanalEventUtils {
/**
* 找出一个最小的position位置，相等的情况返回position1
*/
public static LogPosition min(LogPosition position1, LogPosition position2) {
if (position1.getIdentity().equals(position2.getIdentity())) {
// 首先根据文件进行比较
if (position1.getPostion().getJournalName().compareTo(position2.getPostion().getJournalName()) > 0) {
return position2;
} else if (position1.getPostion().getJournalName().compareTo(position2.getPostion().getJournalName()) < 0) {
return position1;
} else {
// 根据offest进行比较
if (position1.getPostion().getPosition() > position2.getPostion().getPosition()) {
return position2;
} else {
return position1;
}
}
} else {
// 不同的主备库，根据时间进行比较
if (position1.getPostion().getTimestamp() > position2.getPostion().getTimestamp()) {
return position2;
} else {
return position1;
}
}
}
/**
* 根据entry创建对应的Position对象
*/
public static LogPosition createPosition(Event event) {
EntryPosition position = new EntryPosition();
position.setJournalName(event.getEntry().getHeader().getLogfileName());
position.setPosition(event.getEntry().getHeader().getLogfileOffset());
position.setTimestamp(event.getEntry().getHeader().getExecuteTime());
LogPosition logPosition = new LogPosition();
logPosition.setPostion(position);
logPosition.setIdentity(event.getLogIdentity());
return logPosition;
}
/**
* 根据entry创建对应的Position对象
*/
public static LogPosition createPosition(Event event, boolean included) {
EntryPosition position = new EntryPosition();
position.setJournalName(event.getEntry().getHeader().getLogfileName());
position.setPosition(event.getEntry().getHeader().getLogfileOffset());
position.setTimestamp(event.getEntry().getHeader().getExecuteTime());
position.setIncluded(included);
LogPosition logPosition = new LogPosition();
logPosition.setPostion(position);
logPosition.setIdentity(event.getLogIdentity());
return logPosition;
}
/**
* 判断当前的entry和position是否相同
*/
public static boolean checkPosition(Event event, LogPosition logPosition) {
EntryPosition position = logPosition.getPostion();
CanalEntry.Entry entry = event.getEntry();
boolean result = position.getTimestamp().equals(entry.getHeader().getExecuteTime());
boolean exactely = (StringUtils.isBlank(position.getJournalName()) && position.getPosition() == null);
if (!exactely) {// 精确匹配
result &= StringUtils.equals(entry.getHeader().getLogfileName(), position.getJournalName());
result &= position.getPosition().equals(entry.getHeader().getLogfileOffset());
}
return result;
}
}

566
store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java Normal file
View File

@@ -0,0 +1,566 @@
package com.alibaba.otter.canal.store.memory;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.position.LogPosition;
import com.alibaba.otter.canal.protocol.position.Position;
import com.alibaba.otter.canal.protocol.position.PositionRange;
import com.alibaba.otter.canal.store.AbstractCanalStoreScavenge;
import com.alibaba.otter.canal.store.CanalEventStore;
import com.alibaba.otter.canal.store.CanalStoreException;
import com.alibaba.otter.canal.store.CanalStoreScavenge;
import com.alibaba.otter.canal.store.helper.CanalEventUtils;
import com.alibaba.otter.canal.store.model.BatchMode;
import com.alibaba.otter.canal.store.model.Event;
import com.alibaba.otter.canal.store.model.Events;
/**
* 基于内存buffer构建内存memory store
*
* <pre>
* 变更记录：
* 1. 新增BatchMode类型，支持按内存大小获取批次数据，内存大小更加可控.
* a. put操作，会首先根据bufferSize进行控制，然后再进行bufferSize * bufferMemUnit进行控制. 因存储的内容是以Event，如果纯依赖于memsize进行控制，会导致RingBuffer出现动态伸缩
* </pre>
*
* @author jianghang 2012-6-20 上午09:46:31
* @version 1.0.0
*/
public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge implements CanalEventStore<Event>, CanalStoreScavenge {
private static final long INIT_SQEUENCE = -1;
private int bufferSize = 16 * 1024;
private int bufferMemUnit = 1024; // memsize的单位，默认为1kb大小
private int indexMask;
private Event[] entries;
// 记录下put/get/ack操作的三个下标
private AtomicLong putSequence = new AtomicLong(INIT_SQEUENCE); // 代表当前put操作最后一次写操作发生的位置
private AtomicLong getSequence = new AtomicLong(INIT_SQEUENCE); // 代表当前get操作读取的最后一条的位置
private AtomicLong ackSequence = new AtomicLong(INIT_SQEUENCE); // 代表当前ack操作的最后一条的位置
// 记录下put/get/ack操作的三个memsize大小
private AtomicLong putMemSize = new AtomicLong(0);
private AtomicLong getMemSize = new AtomicLong(0);
private AtomicLong ackMemSize = new AtomicLong(0);
// 阻塞put/get操作控制信号
private ReentrantLock lock = new ReentrantLock();
private Condition notFull = lock.newCondition();
private Condition notEmpty = lock.newCondition();
private BatchMode batchMode = BatchMode.ITEMSIZE; // 默认为内存大小模式
private boolean ddlIsolation = false;
public MemoryEventStoreWithBuffer(){
}
public MemoryEventStoreWithBuffer(BatchMode batchMode){
this.batchMode = batchMode;
}
public void start() throws CanalStoreException {
super.start();
if (Integer.bitCount(bufferSize) != 1) {
throw new IllegalArgumentException("bufferSize must be a power of 2");
}
indexMask = bufferSize - 1;
entries = new Event[bufferSize];
}
public void stop() throws CanalStoreException {
super.stop();
cleanAll();
}
public void put(List<Event> data) throws InterruptedException, CanalStoreException {
if (data == null || data.isEmpty()) {
return;
}
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
try {
while (!checkFreeSlotAt(putSequence.get() + data.size())) { // 检查是否有空位
notFull.await(); // wait until not full
}
} catch (InterruptedException ie) {
notFull.signal(); // propagate to non-interrupted thread
throw ie;
}
doPut(data);
if (Thread.interrupted()) {
throw new InterruptedException();
}
} finally {
lock.unlock();
}
}
public boolean put(List<Event> data, long timeout, TimeUnit unit) throws InterruptedException, CanalStoreException {
if (data == null || data.isEmpty()) {
return true;
}
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
if (checkFreeSlotAt(putSequence.get() + data.size())) {
doPut(data);
return true;
}
if (nanos <= 0) {
return false;
}
try {
nanos = notFull.awaitNanos(nanos);
} catch (InterruptedException ie) {
notFull.signal(); // propagate to non-interrupted thread
throw ie;
}
}
} finally {
lock.unlock();
}
}
public boolean tryPut(List<Event> data) throws CanalStoreException {
if (data == null || data.isEmpty()) {
return true;
}
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (!checkFreeSlotAt(putSequence.get() + data.size())) {
return false;
} else {
doPut(data);
return true;
}
} finally {
lock.unlock();
}
}
public void put(Event data) throws InterruptedException, CanalStoreException {
put(Arrays.asList(data));
}
public boolean put(Event data, long timeout, TimeUnit unit) throws InterruptedException, CanalStoreException {
return put(Arrays.asList(data), timeout, unit);
}
public boolean tryPut(Event data) throws CanalStoreException {
return tryPut(Arrays.asList(data));
}
/**
* 执行具体的put操作
*/
private void doPut(List<Event> data) {
long current = putSequence.get();
long end = current + data.size();
// 先写数据，再更新对应的cursor,并发度高的情况，putSequence会被get请求可见，拿出了ringbuffer中的老的Entry值
for (long next = current + 1; next <= end; next++) {
entries[getIndex(next)] = data.get((int) (next - current - 1));
}
putSequence.set(end);
// 记录一下gets memsize信息，方便快速检索
if (batchMode.isMemSize()) {
long size = 0;
for (Event event : data) {
size += calculateSize(event);
}
putMemSize.getAndAdd(size);
}
// tell other threads that store is not empty
notEmpty.signal();
}
public Events<Event> get(Position start, int batchSize) throws InterruptedException, CanalStoreException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
try {
while (!checkUnGetSlotAt((LogPosition) start, batchSize))
notEmpty.await();
} catch (InterruptedException ie) {
notEmpty.signal(); // propagate to non-interrupted thread
throw ie;
}
return doGet(start, batchSize);
} finally {
lock.unlock();
}
}
public Events<Event> get(Position start, int batchSize, long timeout, TimeUnit unit) throws InterruptedException,
CanalStoreException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
if (checkUnGetSlotAt((LogPosition) start, batchSize)) {
return doGet(start, batchSize);
}
if (nanos <= 0) {
// 如果时间到了，有多少取多少
return doGet(start, batchSize);
}
try {
nanos = notEmpty.awaitNanos(nanos);
} catch (InterruptedException ie) {
notEmpty.signal(); // propagate to non-interrupted thread
throw ie;
}
}
} finally {
lock.unlock();
}
}
public Events<Event> tryGet(Position start, int batchSize) throws CanalStoreException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return doGet(start, batchSize);
} finally {
lock.unlock();
}
}
private Events<Event> doGet(Position start, int batchSize) throws CanalStoreException {
LogPosition startPosition = (LogPosition) start;
long current = getSequence.get();
long maxAbleSequence = putSequence.get();
long next = current;
long end = current;
// 如果startPosition为null，说明是第一次，默认+1处理
if (startPosition == null || !startPosition.getPostion().isIncluded()) { // 第一次订阅之后，需要包含一下start位置，防止丢失第一条记录
next = next + 1;
}
if (current >= maxAbleSequence) {
return new Events<Event>();
}
Events<Event> result = new Events<Event>();
List<Event> entrys = result.getEvents();
long memsize = 0;
if (batchMode.isItemSize()) {
end = (next + batchSize - 1) < maxAbleSequence ? (next + batchSize - 1) : maxAbleSequence;
// 提取数据并返回
for (; next <= end; next++) {
Event event = entries[getIndex(next)];
if (ddlIsolation && isDdl(event.getEntry().getHeader().getEventType())) {
// 如果是ddl隔离，直接返回
if (entrys.size() == 0) {
entrys.add(event);// 如果没有DML事件，加入当前的DDL事件
end = next; // 更新end为当前
} else {
// 如果之前已经有DML事件，直接返回了，因为不包含当前next这记录，需要回退一个位置
end = next - 1; // next-1一定大于current，不需要判断
}
break;
} else {
entrys.add(event);
}
}
} else {
long maxMemSize = batchSize * bufferMemUnit;
for (; memsize <= maxMemSize && next <= maxAbleSequence; next++) {
// 永远保证可以取出第一条的记录，避免死锁
Event event = entries[getIndex(next)];
if (ddlIsolation && isDdl(event.getEntry().getHeader().getEventType())) {
// 如果是ddl隔离，直接返回
if (entrys.size() == 0) {
entrys.add(event);// 如果没有DML事件，加入当前的DDL事件
end = next; // 更新end为当前
} else {
// 如果之前已经有DML事件，直接返回了，因为不包含当前next这记录，需要回退一个位置
end = next - 1; // next-1一定大于current，不需要判断
}
break;
} else {
entrys.add(event);
memsize += calculateSize(event);
end = next;// 记录end位点
}
}
}
PositionRange<LogPosition> range = new PositionRange<LogPosition>();
result.setPositionRange(range);
range.setStart(CanalEventUtils.createPosition(entrys.get(0)));
range.setEnd(CanalEventUtils.createPosition(entrys.get(result.getEvents().size() - 1)));
// 记录一下是否存在可以被ack的点
for (int i = entrys.size() - 1; i >= 0; i--) {
Event event = entrys.get(i);
if (CanalEntry.EntryType.TRANSACTIONBEGIN == event.getEntry().getEntryType()
|| CanalEntry.EntryType.TRANSACTIONEND == event.getEntry().getEntryType()
|| isDdl(event.getEntry().getHeader().getEventType())) {
// 将事务头/尾设置可被为ack的点
range.setAck(CanalEventUtils.createPosition(event));
break;
}
}
if (getSequence.compareAndSet(current, end)) {
getMemSize.addAndGet(memsize);
notFull.signal();
return result;
} else {
return new Events<Event>();
}
}
public LogPosition getFirstPosition() throws CanalStoreException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
long firstSeqeuence = ackSequence.get();
if (firstSeqeuence == INIT_SQEUENCE && firstSeqeuence < putSequence.get()) {
// 没有ack过数据
Event event = entries[getIndex(firstSeqeuence + 1)]; // 最后一次ack为-1，需要移动到下一条,included
// = false
return CanalEventUtils.createPosition(event, false);
} else if (firstSeqeuence > INIT_SQEUENCE && firstSeqeuence < putSequence.get()) {
// ack未追上put操作
Event event = entries[getIndex(firstSeqeuence + 1)]; // 最后一次ack的位置数据
// + 1
return CanalEventUtils.createPosition(event, true);
} else if (firstSeqeuence > INIT_SQEUENCE && firstSeqeuence == putSequence.get()) {
// 已经追上，store中没有数据
Event event = entries[getIndex(firstSeqeuence)]; // 最后一次ack的位置数据，和last为同一条，included
// = false
return CanalEventUtils.createPosition(event, false);
} else {
// 没有任何数据
return null;
}
} finally {
lock.unlock();
}
}
public LogPosition getLatestPosition() throws CanalStoreException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
long latestSequence = putSequence.get();
if (latestSequence > INIT_SQEUENCE && latestSequence != ackSequence.get()) {
Event event = entries[(int) putSequence.get() & indexMask]; // 最后一次写入的数据，最后一条未消费的数据
return CanalEventUtils.createPosition(event, true);
} else if (latestSequence > INIT_SQEUENCE && latestSequence == ackSequence.get()) {
// ack已经追上了put操作
Event event = entries[(int) putSequence.get() & indexMask]; // 最后一次写入的数据，included
// =
// false
return CanalEventUtils.createPosition(event, false);
} else {
// 没有任何数据
return null;
}
} finally {
lock.unlock();
}
}
public void ack(Position position) throws CanalStoreException {
cleanUntil(position);
}
public void cleanUntil(Position position) throws CanalStoreException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
long sequence = ackSequence.get();
long maxSequence = getSequence.get();
boolean hasMatch = false;
long memsize = 0;
for (long next = sequence + 1; next <= maxSequence; next++) {
Event event = entries[getIndex(next)];
memsize += calculateSize(event);
boolean match = CanalEventUtils.checkPosition(event, (LogPosition) position);
if (match) {// 找到对应的position，更新ack seq
hasMatch = true;
if (batchMode.isMemSize()) {
ackMemSize.addAndGet(memsize);
// 尝试清空buffer中的内存，将ack之前的内存全部释放掉
for (long index = sequence + 1; index < next; index++) {
entries[getIndex(index)] = null;// 设置为null
}
}
if (ackSequence.compareAndSet(sequence, next)) {// 避免并发ack
notFull.signal();
return;
}
}
}
if (!hasMatch) {// 找不到对应需要ack的position
throw new CanalStoreException("no match ack position" + position.toString());
}
} finally {
lock.unlock();
}
}
public void rollback() throws CanalStoreException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
getSequence.set(ackSequence.get());
getMemSize.set(ackMemSize.get());
} finally {
lock.unlock();
}
}
public void cleanAll() throws CanalStoreException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
putSequence.set(INIT_SQEUENCE);
getSequence.set(INIT_SQEUENCE);
ackSequence.set(INIT_SQEUENCE);
putMemSize.set(0);
getMemSize.set(0);
ackMemSize.set(0);
entries = null;
// for (int i = 0; i < entries.length; i++) {
// entries[i] = null;
// }
} finally {
lock.unlock();
}
}
// =================== helper method =================
private long getMinimumGetOrAck() {
long get = getSequence.get();
long ack = ackSequence.get();
return ack <= get ? ack : get;
}
/**
* 查询是否有空位
*/
private boolean checkFreeSlotAt(final long sequence) {
final long wrapPoint = sequence - bufferSize;
final long minPoint = getMinimumGetOrAck();
if (wrapPoint > minPoint) { // 刚好追上一轮
return false;
} else {
// 在bufferSize模式上，再增加memSize控制
if (batchMode.isMemSize()) {
final long memsize = putMemSize.get() - ackMemSize.get();
if (memsize < bufferSize * bufferMemUnit) {
return true;
} else {
return false;
}
} else {
return true;
}
}
}
/**
* 检查是否存在需要get的数据,并且数量>=batchSize
*/
private boolean checkUnGetSlotAt(LogPosition startPosition, int batchSize) {
if (batchMode.isItemSize()) {
long current = getSequence.get();
long maxAbleSequence = putSequence.get();
long next = current;
if (startPosition == null || !startPosition.getPostion().isIncluded()) { // 第一次订阅之后，需要包含一下start位置，防止丢失第一条记录
next = next + 1;// 少一条数据
}
if (current < maxAbleSequence && next + batchSize - 1 <= maxAbleSequence) {
return true;
} else {
return false;
}
} else {
// 处理内存大小判断
long currentSize = getMemSize.get();
long maxAbleSize = putMemSize.get();
if (maxAbleSize - currentSize >= batchSize * bufferMemUnit) {
return true;
} else {
return false;
}
}
}
private long calculateSize(Event event) {
// 直接返回binlog中的事件大小
return event.getEntry().getHeader().getEventLength();
}
private int getIndex(long sequcnce) {
return (int) sequcnce & indexMask;
}
private boolean isDdl(EventType type) {
return type == EventType.ALTER || type == EventType.CREATE || type == EventType.ERASE
|| type == EventType.RENAME || type == EventType.TRUNCATE || type == EventType.CINDEX
|| type == EventType.DINDEX;
}
// ================ setter / getter ==================
public void setBufferSize(int bufferSize) {
this.bufferSize = bufferSize;
}
public void setBufferMemUnit(int bufferMemUnit) {
this.bufferMemUnit = bufferMemUnit;
}
public void setBatchMode(BatchMode batchMode) {
this.batchMode = batchMode;
}
public void setDdlIsolation(boolean ddlIsolation) {
this.ddlIsolation = ddlIsolation;
}
}

24
store/src/main/java/com/alibaba/otter/canal/store/model/BatchMode.java Normal file
View File

@@ -0,0 +1,24 @@
package com.alibaba.otter.canal.store.model;
/**
* 批处理模式
*
* @author jianghang 2013-3-18 上午11:51:15
* @version 1.0.3
*/
public enum BatchMode {
/** 对象数量 */
ITEMSIZE,
/** 内存大小 */
MEMSIZE;
public boolean isItemSize() {
return this == BatchMode.ITEMSIZE;
}
public boolean isMemSize() {
return this == BatchMode.MEMSIZE;
}
}

50
store/src/main/java/com/alibaba/otter/canal/store/model/Event.java Normal file
View File

@@ -0,0 +1,50 @@
package com.alibaba.otter.canal.store.model;
import java.io.Serializable;
import org.apache.commons.lang.builder.ToStringBuilder;
import com.alibaba.otter.canal.common.utils.CanalToStringStyle;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.position.LogIdentity;
/**
* store存储数据对象
*
* @author jianghang 2012-7-13 下午03:03:03
*/
public class Event implements Serializable {
private static final long serialVersionUID = 1333330351758762739L;
private LogIdentity logIdentity; // 记录数据产生的来源
private CanalEntry.Entry entry;
public Event(){
}
public Event(LogIdentity logIdentity, CanalEntry.Entry entry){
this.logIdentity = logIdentity;
this.entry = entry;
}
public LogIdentity getLogIdentity() {
return logIdentity;
}
public void setLogIdentity(LogIdentity logIdentity) {
this.logIdentity = logIdentity;
}
public CanalEntry.Entry getEntry() {
return entry;
}
public void setEntry(CanalEntry.Entry entry) {
this.entry = entry;
}
public String toString() {
return ToStringBuilder.reflectionToString(this, CanalToStringStyle.DEFAULT_STYLE);
}
}

44
store/src/main/java/com/alibaba/otter/canal/store/model/Events.java Normal file
View File

@@ -0,0 +1,44 @@
package com.alibaba.otter.canal.store.model;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.lang.builder.ToStringBuilder;
import com.alibaba.otter.canal.common.utils.CanalToStringStyle;
import com.alibaba.otter.canal.protocol.position.PositionRange;
/**
* 代表一组数据对象的集合
*
* @author jianghang 2012-6-14 下午09:07:41
* @version 1.0.0
*/
public class Events<EVENT> implements Serializable {
private static final long serialVersionUID = -7337454954300706044L;
private PositionRange positionRange = new PositionRange();
private List<EVENT> events = new ArrayList<EVENT>();
public List<EVENT> getEvents() {
return events;
}
public void setEvents(List<EVENT> events) {
this.events = events;
}
public PositionRange getPositionRange() {
return positionRange;
}
public void setPositionRange(PositionRange positionRange) {
this.positionRange = positionRange;
}
public String toString() {
return ToStringBuilder.reflectionToString(this, CanalToStringStyle.DEFAULT_STYLE);
}
}

49
store/src/test/java/com/alibaba/otter/cancel/store/memory/buffer/MemoryEventStoreBase.java Normal file
View File

@@ -0,0 +1,49 @@
package com.alibaba.otter.cancel.store.memory.buffer;
import java.net.InetSocketAddress;
import org.junit.Assert;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.Header;
import com.alibaba.otter.canal.protocol.position.LogIdentity;
import com.alibaba.otter.canal.store.model.Event;
public class MemoryEventStoreBase {
private static final String MYSQL_ADDRESS = "127.0.0.1";
protected void sleep(Long time) {
try {
Thread.sleep(time);
} catch (InterruptedException e) {
Assert.fail();
}
}
protected Event buildEvent(String binlogFile, long offset, long timestamp) {
Header.Builder headerBuilder = Header.newBuilder();
headerBuilder.setLogfileName(binlogFile);
headerBuilder.setLogfileOffset(offset);
headerBuilder.setExecuteTime(timestamp);
headerBuilder.setEventLength(1024);
Entry.Builder entryBuilder = Entry.newBuilder();
entryBuilder.setHeader(headerBuilder.build());
Entry entry = entryBuilder.build();
return new Event(new LogIdentity(new InetSocketAddress(MYSQL_ADDRESS, 3306), 1234L), entry);
}
protected Event buildEvent(String binlogFile, long offset, long timestamp, long eventLenght) {
Header.Builder headerBuilder = Header.newBuilder();
headerBuilder.setLogfileName(binlogFile);
headerBuilder.setLogfileOffset(offset);
headerBuilder.setExecuteTime(timestamp);
headerBuilder.setEventLength(eventLenght);
Entry.Builder entryBuilder = Entry.newBuilder();
entryBuilder.setHeader(headerBuilder.build());
Entry entry = entryBuilder.build();
return new Event(new LogIdentity(new InetSocketAddress(MYSQL_ADDRESS, 3306), 1234L), entry);
}
}

317
store/src/test/java/com/alibaba/otter/cancel/store/memory/buffer/MemoryEventStoreMemBatchTest.java Normal file
View File

@@ -0,0 +1,317 @@
package com.alibaba.otter.cancel.store.memory.buffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.junit.Assert;
import org.junit.Test;
import com.alibaba.otter.canal.protocol.position.Position;
import com.alibaba.otter.canal.store.CanalStoreException;
import com.alibaba.otter.canal.store.helper.CanalEventUtils;
import com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer;
import com.alibaba.otter.canal.store.model.BatchMode;
import com.alibaba.otter.canal.store.model.Event;
import com.alibaba.otter.canal.store.model.Events;
public class MemoryEventStoreMemBatchTest extends MemoryEventStoreBase {
@Test
public void testOnePut() {
MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.setBatchMode(BatchMode.MEMSIZE);
eventStore.start();
// 尝试阻塞
try {
eventStore.put(buildEvent("1", 1L, 1L, 1024));
} catch (Exception e) {
Assert.fail(e.getMessage());
}
// 尝试阻塞+超时
boolean result = false;
try {
result = eventStore.put(buildEvent("1", 1L, 1L), 1000L, TimeUnit.MILLISECONDS);
Assert.assertTrue(result);
} catch (Exception e) {
Assert.fail(e.getMessage());
}
// 尝试
result = eventStore.tryPut(buildEvent("1", 1L, 1L));
Assert.assertTrue(result);
eventStore.stop();
}
@Test
public void testOnePutExceedLimit() {
MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.setBufferSize(1);
eventStore.setBatchMode(BatchMode.MEMSIZE);
eventStore.start();
// 尝试阻塞
try {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L, 1025));// 只有一条记录，第一条超过也允许放入
Assert.assertTrue(result);
} catch (Exception e) {
Assert.fail(e.getMessage());
}
eventStore.stop();
}
@Test
public void testFullPut() {
int bufferSize = 16;
MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.setBufferSize(bufferSize);
eventStore.setBatchMode(BatchMode.MEMSIZE);
eventStore.start();
for (int i = 0; i < bufferSize; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
Assert.assertTrue(result);
}
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + bufferSize));
Assert.assertFalse(result);
try {
result = eventStore.put(buildEvent("1", 1L, 1L + bufferSize), 1000L, TimeUnit.MILLISECONDS);
} catch (CanalStoreException e) {
Assert.fail(e.getMessage());
} catch (InterruptedException e) {
Assert.fail(e.getMessage());
}
Assert.assertFalse(result);
eventStore.stop();
}
@Test
public void testOnePutOneGet() {
MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.setBatchMode(BatchMode.MEMSIZE);
eventStore.start();
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L));
Assert.assertTrue(result);
Position position = eventStore.getFirstPosition();
Events<Event> entrys = eventStore.tryGet(position, 1);
Assert.assertTrue(entrys.getEvents().size() == 1);
Assert.assertEquals(position, entrys.getPositionRange().getStart());
Assert.assertEquals(position, entrys.getPositionRange().getEnd());
eventStore.stop();
}
@Test
public void testFullPutBatchGet() {
int bufferSize = 16;
MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.setBufferSize(bufferSize);
eventStore.setBatchMode(BatchMode.MEMSIZE);
eventStore.start();
for (int i = 0; i < bufferSize; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
sleep(100L);
Assert.assertTrue(result);
}
Position first = eventStore.getFirstPosition();
Position lastest = eventStore.getLatestPosition();
Assert.assertEquals(first, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L)));
Assert.assertEquals(lastest, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L + bufferSize - 1)));
System.out.println("start get");
Events<Event> entrys1 = eventStore.tryGet(first, bufferSize);
System.out.println("first get size : " + entrys1.getEvents().size());
Assert.assertTrue(entrys1.getEvents().size() == bufferSize);
Assert.assertEquals(first, entrys1.getPositionRange().getStart());
Assert.assertEquals(lastest, entrys1.getPositionRange().getEnd());
Assert.assertEquals(first, CanalEventUtils.createPosition(entrys1.getEvents().get(0)));
Assert.assertEquals(lastest, CanalEventUtils.createPosition(entrys1.getEvents().get(bufferSize - 1)));
eventStore.stop();
}
@Test
public void testBlockPutOneGet() {
final MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.setBufferSize(16);
eventStore.setBatchMode(BatchMode.MEMSIZE);
eventStore.start();
final int batchSize = 10;
for (int i = 0; i < batchSize; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L));
Assert.assertTrue(result);
}
final Position position = eventStore.getFirstPosition();
try {
Events<Event> entrys = eventStore.get(position, batchSize);
Assert.assertTrue(entrys.getEvents().size() == batchSize);
Assert.assertEquals(position, entrys.getPositionRange().getStart());
Assert.assertEquals(position, entrys.getPositionRange().getEnd());
} catch (CanalStoreException e) {
} catch (InterruptedException e) {
}
ExecutorService executor = Executors.newFixedThreadPool(1);
executor.submit(new Runnable() {
public void run() {
boolean result = false;
try {
eventStore.get(position, batchSize);
} catch (CanalStoreException e) {
} catch (InterruptedException e) {
System.out.println("interrupt occured.");
result = true;
}
Assert.assertTrue(result);
}
});
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
Assert.fail(e.getMessage());
}
executor.shutdownNow();
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
Assert.fail(e.getMessage());
}
eventStore.stop();
}
@Test
public void testRollback() {
int bufferSize = 16;
MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.setBufferSize(bufferSize);
eventStore.setBatchMode(BatchMode.MEMSIZE);
eventStore.start();
for (int i = 0; i < bufferSize / 2; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
sleep(100L);
Assert.assertTrue(result);
}
sleep(50L);
Position first = eventStore.getFirstPosition();
Position lastest = eventStore.getLatestPosition();
Assert.assertEquals(first, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L)));
Assert.assertEquals(lastest, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L + bufferSize / 2 - 1)));
System.out.println("start get");
Events<Event> entrys1 = eventStore.tryGet(first, bufferSize);
System.out.println("first get size : " + entrys1.getEvents().size());
eventStore.rollback();
entrys1 = eventStore.tryGet(first, bufferSize);
System.out.println("after rollback get size : " + entrys1.getEvents().size());
Assert.assertTrue(entrys1.getEvents().size() == bufferSize / 2);
// 继续造数据
for (int i = bufferSize / 2; i < bufferSize; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
sleep(100L);
Assert.assertTrue(result);
}
Events<Event> entrys2 = eventStore.tryGet(entrys1.getPositionRange().getEnd(), bufferSize);
System.out.println("second get size : " + entrys2.getEvents().size());
eventStore.rollback();
entrys2 = eventStore.tryGet(entrys1.getPositionRange().getEnd(), bufferSize);
System.out.println("after rollback get size : " + entrys2.getEvents().size());
Assert.assertTrue(entrys2.getEvents().size() == bufferSize);
first = eventStore.getFirstPosition();
lastest = eventStore.getLatestPosition();
List<Event> entrys = new ArrayList<Event>(entrys2.getEvents());
Assert.assertTrue(entrys.size() == bufferSize);
Assert.assertEquals(first, entrys2.getPositionRange().getStart());
Assert.assertEquals(lastest, entrys2.getPositionRange().getEnd());
Assert.assertEquals(first, CanalEventUtils.createPosition(entrys.get(0)));
Assert.assertEquals(lastest, CanalEventUtils.createPosition(entrys.get(bufferSize - 1)));
eventStore.stop();
}
@Test
public void testAck() {
int bufferSize = 16;
MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.setBufferSize(bufferSize);
eventStore.setBatchMode(BatchMode.MEMSIZE);
eventStore.start();
for (int i = 0; i < bufferSize / 2; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
sleep(100L);
Assert.assertTrue(result);
}
sleep(50L);
Position first = eventStore.getFirstPosition();
Position lastest = eventStore.getLatestPosition();
Assert.assertEquals(first, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L)));
Assert.assertEquals(lastest, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L + bufferSize / 2 - 1)));
System.out.println("start get");
Events<Event> entrys1 = eventStore.tryGet(first, bufferSize);
System.out.println("first get size : " + entrys1.getEvents().size());
eventStore.cleanUntil(entrys1.getPositionRange().getEnd());
sleep(50L);
// 继续造数据
for (int i = bufferSize / 2; i < bufferSize; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
sleep(100L);
Assert.assertTrue(result);
}
Events<Event> entrys2 = eventStore.tryGet(entrys1.getPositionRange().getEnd(), bufferSize);
System.out.println("second get size : " + entrys2.getEvents().size());
eventStore.rollback();
entrys2 = eventStore.tryGet(entrys1.getPositionRange().getEnd(), bufferSize);
System.out.println("after rollback get size : " + entrys2.getEvents().size());
first = eventStore.getFirstPosition();
lastest = eventStore.getLatestPosition();
List<Event> entrys = new ArrayList<Event>(entrys2.getEvents());
Assert.assertEquals(first, entrys2.getPositionRange().getStart());
Assert.assertEquals(lastest, entrys2.getPositionRange().getEnd());
Assert.assertEquals(first, CanalEventUtils.createPosition(entrys.get(0)));
Assert.assertEquals(lastest, CanalEventUtils.createPosition(entrys.get(entrys.size() - 1)));
// 全部ack掉
eventStore.cleanUntil(entrys2.getPositionRange().getEnd());
// 最后就拿不到数据
Events<Event> entrys3 = eventStore.tryGet(entrys1.getPositionRange().getEnd(), bufferSize);
System.out.println("third get size : " + entrys3.getEvents().size());
Assert.assertEquals(0, entrys3.getEvents().size());
eventStore.stop();
}
}

191
store/src/test/java/com/alibaba/otter/cancel/store/memory/buffer/MemoryEventStoreMultiThreadTest.java Normal file
View File

@@ -0,0 +1,191 @@
package com.alibaba.otter.cancel.store.memory.buffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import junit.framework.Assert;
import org.apache.commons.lang.math.RandomUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.util.CollectionUtils;
import com.alibaba.otter.canal.common.utils.BooleanMutex;
import com.alibaba.otter.canal.protocol.position.Position;
import com.alibaba.otter.canal.store.CanalStoreException;
import com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer;
import com.alibaba.otter.canal.store.model.BatchMode;
import com.alibaba.otter.canal.store.model.Event;
import com.alibaba.otter.canal.store.model.Events;
/**
* 多线程的put/get/ack/rollback测试
*
* @author jianghang 2012-6-20 下午02:50:36
* @version 1.0.0
*/
public class MemoryEventStoreMultiThreadTest extends MemoryEventStoreBase {
private ExecutorService executor = Executors.newFixedThreadPool(2); // 1 producer ,1 cousmer
private MemoryEventStoreWithBuffer eventStore;
@Before
public void setUp() {
eventStore = new MemoryEventStoreWithBuffer();
eventStore.setBufferSize(16 * 16);
eventStore.setBatchMode(BatchMode.MEMSIZE);
eventStore.start();
}
@After
public void tearDown() {
eventStore.stop();
}
@Test
public void test() {
CountDownLatch latch = new CountDownLatch(1);
BooleanMutex mutex = new BooleanMutex(true);
Producer producer = new Producer(mutex, 10);
Cosumer cosumer = new Cosumer(latch, 20, 50);
executor.submit(producer);
executor.submit(cosumer);
try {
Thread.sleep(30 * 1000L);
} catch (InterruptedException e) {
}
mutex.set(false);
try {
latch.await();
} catch (InterruptedException e) {
}
executor.shutdown();
List<Long> result = cosumer.getResult();
Long last = -1L;
for (Long offest : result) {
Assert.assertTrue(last + 1 == offest);// 取出来的数据一定是递增的
last = offest;
}
}
class Producer implements Runnable {
private BooleanMutex mutex;
private int freq;
public Producer(BooleanMutex mutex, int freq){
this.mutex = mutex;
this.freq = freq;
}
public void run() {
long offest = 0;
while (true) {
try {
mutex.get();
Thread.sleep(RandomUtils.nextInt(freq));
} catch (InterruptedException e) {
return;
}
Event event = buildEvent("1", offest++, 1L);
try {
Thread.sleep(RandomUtils.nextInt(freq));
} catch (InterruptedException e) {
return;
}
try {
eventStore.put(event);
} catch (CanalStoreException e) {
} catch (InterruptedException e) {
}
}
}
}
class Cosumer implements Runnable {
private CountDownLatch latch;
private int freq;
private int batchSize;
private List<Long> result = new ArrayList<Long>();
public Cosumer(CountDownLatch latch, int freq, int batchSize){
this.latch = latch;
this.freq = freq;
this.batchSize = batchSize;
}
public void run() {
Position first = eventStore.getFirstPosition();
while (first == null) {
try {
Thread.sleep(RandomUtils.nextInt(freq));
} catch (InterruptedException e) {
latch.countDown();
return;
}
first = eventStore.getFirstPosition();
}
int ackCount = 0;
int emptyCount = 0;
while (emptyCount < 10) {
try {
Thread.sleep(RandomUtils.nextInt(freq));
} catch (InterruptedException e) {
}
try {
Events<Event> entrys = eventStore.get(first, batchSize, 1000L, TimeUnit.MILLISECONDS);
// Events<Event> entrys = eventStore.tryGet(first, batchSize);
if (!CollectionUtils.isEmpty(entrys.getEvents())) {
if (entrys.getEvents().size() != batchSize) {
System.out.println("get size:" + entrys.getEvents().size() + " with not full batchSize:"
+ batchSize);
}
first = entrys.getPositionRange().getEnd();
for (Event event : entrys.getEvents()) {
this.result.add(event.getEntry().getHeader().getLogfileOffset());
}
emptyCount = 0;
System.out.println("offest : "
+ entrys.getEvents().get(0).getEntry().getHeader().getLogfileOffset()
+ " , count :" + entrys.getEvents().size());
ackCount++;
if (ackCount == 1) {
eventStore.cleanUntil(entrys.getPositionRange().getEnd());
System.out.println("first position : " + eventStore.getFirstPosition());
ackCount = 0;
}
} else {
emptyCount++;
System.out.println("empty events for " + emptyCount);
}
} catch (Exception e) {
e.printStackTrace();
}
}
latch.countDown();
}
public List<Long> getResult() {
return result;
}
}
}

177
store/src/test/java/com/alibaba/otter/cancel/store/memory/buffer/MemoryEventStorePutAndGetTest.java Normal file
View File

@@ -0,0 +1,177 @@
package com.alibaba.otter.cancel.store.memory.buffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.junit.Assert;
import org.junit.Test;
import com.alibaba.otter.canal.protocol.position.Position;
import com.alibaba.otter.canal.store.CanalStoreException;
import com.alibaba.otter.canal.store.helper.CanalEventUtils;
import com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer;
import com.alibaba.otter.canal.store.model.Event;
import com.alibaba.otter.canal.store.model.Events;
/**
* 测试普通的put / get操作
*
* @author jianghang 2012-6-19 下午09:50:08
* @version 1.0.0
*/
public class MemoryEventStorePutAndGetTest extends MemoryEventStoreBase {
@Test
public void testOnePut() {
MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.start();
// 尝试阻塞
try {
eventStore.put(buildEvent("1", 1L, 1L));
} catch (Exception e) {
Assert.fail(e.getMessage());
}
// 尝试阻塞+超时
boolean result = false;
try {
result = eventStore.put(buildEvent("1", 1L, 1L), 1000L, TimeUnit.MILLISECONDS);
Assert.assertTrue(result);
} catch (Exception e) {
Assert.fail(e.getMessage());
}
// 尝试
result = eventStore.tryPut(buildEvent("1", 1L, 1L));
Assert.assertTrue(result);
eventStore.stop();
}
@Test
public void testFullPut() {
int bufferSize = 16;
MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.setBufferSize(bufferSize);
eventStore.start();
for (int i = 0; i < bufferSize; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
Assert.assertTrue(result);
}
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + bufferSize));
Assert.assertFalse(result);
try {
result = eventStore.put(buildEvent("1", 1L, 1L + bufferSize), 1000L, TimeUnit.MILLISECONDS);
} catch (CanalStoreException e) {
Assert.fail(e.getMessage());
} catch (InterruptedException e) {
Assert.fail(e.getMessage());
}
Assert.assertFalse(result);
eventStore.stop();
}
@Test
public void testOnePutOneGet() {
MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.start();
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L));
Assert.assertTrue(result);
Position position = eventStore.getFirstPosition();
Events<Event> entrys = eventStore.tryGet(position, 1);
Assert.assertTrue(entrys.getEvents().size() == 1);
Assert.assertEquals(position, entrys.getPositionRange().getStart());
Assert.assertEquals(position, entrys.getPositionRange().getEnd());
eventStore.stop();
}
@Test
public void testFullPutBatchGet() {
int bufferSize = 16;
MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.setBufferSize(bufferSize);
eventStore.start();
for (int i = 0; i < bufferSize; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
sleep(100L);
Assert.assertTrue(result);
}
Position first = eventStore.getFirstPosition();
Position lastest = eventStore.getLatestPosition();
Assert.assertEquals(first, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L)));
Assert.assertEquals(lastest, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L + bufferSize - 1)));
System.out.println("start get");
Events<Event> entrys1 = eventStore.tryGet(first, bufferSize);
System.out.println("first get size : " + entrys1.getEvents().size());
Assert.assertTrue(entrys1.getEvents().size() == bufferSize);
Assert.assertEquals(first, entrys1.getPositionRange().getStart());
Assert.assertEquals(lastest, entrys1.getPositionRange().getEnd());
Assert.assertEquals(first, CanalEventUtils.createPosition(entrys1.getEvents().get(0)));
Assert.assertEquals(lastest, CanalEventUtils.createPosition(entrys1.getEvents().get(bufferSize - 1)));
eventStore.stop();
}
@Test
public void testBlockPutOneGet() {
final MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.start();
final int batchSize = 10;
for (int i = 0; i < batchSize; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L));
Assert.assertTrue(result);
}
final Position position = eventStore.getFirstPosition();
try {
Events<Event> entrys = eventStore.get(position, batchSize);
Assert.assertTrue(entrys.getEvents().size() == batchSize);
Assert.assertEquals(position, entrys.getPositionRange().getStart());
Assert.assertEquals(position, entrys.getPositionRange().getEnd());
} catch (CanalStoreException e) {
} catch (InterruptedException e) {
}
ExecutorService executor = Executors.newFixedThreadPool(1);
executor.submit(new Runnable() {
public void run() {
boolean result = false;
try {
eventStore.get(position, batchSize);
} catch (CanalStoreException e) {
} catch (InterruptedException e) {
System.out.println("interrupt occured.");
result = true;
}
Assert.assertTrue(result);
}
});
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
Assert.fail(e.getMessage());
}
executor.shutdownNow();
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
Assert.fail(e.getMessage());
}
eventStore.stop();
}
}

141
store/src/test/java/com/alibaba/otter/cancel/store/memory/buffer/MemoryEventStoreRollbackAndAckTest.java Normal file
View File

@@ -0,0 +1,141 @@
package com.alibaba.otter.cancel.store.memory.buffer;
import java.util.ArrayList;
import java.util.List;
import org.junit.Assert;
import org.junit.Test;
import com.alibaba.otter.canal.protocol.position.Position;
import com.alibaba.otter.canal.store.helper.CanalEventUtils;
import com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer;
import com.alibaba.otter.canal.store.model.Event;
import com.alibaba.otter.canal.store.model.Events;
/**
* 测试下rollback / ack的操作
*
* @author jianghang 2012-6-19 下午09:49:28
* @version 1.0.0
*/
public class MemoryEventStoreRollbackAndAckTest extends MemoryEventStoreBase {
@Test
public void testRollback() {
int bufferSize = 16;
MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.setBufferSize(bufferSize);
eventStore.start();
for (int i = 0; i < bufferSize / 2; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
sleep(100L);
Assert.assertTrue(result);
}
sleep(50L);
Position first = eventStore.getFirstPosition();
Position lastest = eventStore.getLatestPosition();
Assert.assertEquals(first, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L)));
Assert.assertEquals(lastest, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L + bufferSize / 2 - 1)));
System.out.println("start get");
Events<Event> entrys1 = eventStore.tryGet(first, bufferSize);
System.out.println("first get size : " + entrys1.getEvents().size());
eventStore.rollback();
entrys1 = eventStore.tryGet(first, bufferSize);
System.out.println("after rollback get size : " + entrys1.getEvents().size());
Assert.assertTrue(entrys1.getEvents().size() == bufferSize / 2);
// 继续造数据
for (int i = bufferSize / 2; i < bufferSize; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
sleep(100L);
Assert.assertTrue(result);
}
Events<Event> entrys2 = eventStore.tryGet(entrys1.getPositionRange().getEnd(), bufferSize);
System.out.println("second get size : " + entrys2.getEvents().size());
eventStore.rollback();
entrys2 = eventStore.tryGet(entrys1.getPositionRange().getEnd(), bufferSize);
System.out.println("after rollback get size : " + entrys2.getEvents().size());
Assert.assertTrue(entrys2.getEvents().size() == bufferSize);
first = eventStore.getFirstPosition();
lastest = eventStore.getLatestPosition();
List<Event> entrys = new ArrayList<Event>(entrys2.getEvents());
Assert.assertTrue(entrys.size() == bufferSize);
Assert.assertEquals(first, entrys2.getPositionRange().getStart());
Assert.assertEquals(lastest, entrys2.getPositionRange().getEnd());
Assert.assertEquals(first, CanalEventUtils.createPosition(entrys.get(0)));
Assert.assertEquals(lastest, CanalEventUtils.createPosition(entrys.get(bufferSize - 1)));
eventStore.stop();
}
@Test
public void testAck() {
int bufferSize = 16;
MemoryEventStoreWithBuffer eventStore = new MemoryEventStoreWithBuffer();
eventStore.setBufferSize(bufferSize);
eventStore.start();
for (int i = 0; i < bufferSize / 2; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
sleep(100L);
Assert.assertTrue(result);
}
sleep(50L);
Position first = eventStore.getFirstPosition();
Position lastest = eventStore.getLatestPosition();
Assert.assertEquals(first, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L)));
Assert.assertEquals(lastest, CanalEventUtils.createPosition(buildEvent("1", 1L, 1L + bufferSize / 2 - 1)));
System.out.println("start get");
Events<Event> entrys1 = eventStore.tryGet(first, bufferSize);
System.out.println("first get size : " + entrys1.getEvents().size());
eventStore.cleanUntil(entrys1.getPositionRange().getEnd());
sleep(50L);
// 继续造数据
for (int i = bufferSize / 2; i < bufferSize; i++) {
boolean result = eventStore.tryPut(buildEvent("1", 1L, 1L + i));
sleep(100L);
Assert.assertTrue(result);
}
Events<Event> entrys2 = eventStore.tryGet(entrys1.getPositionRange().getEnd(), bufferSize);
System.out.println("second get size : " + entrys2.getEvents().size());
eventStore.rollback();
entrys2 = eventStore.tryGet(entrys1.getPositionRange().getEnd(), bufferSize);
System.out.println("after rollback get size : " + entrys2.getEvents().size());
first = eventStore.getFirstPosition();
lastest = eventStore.getLatestPosition();
List<Event> entrys = new ArrayList<Event>(entrys2.getEvents());
Assert.assertEquals(first, entrys2.getPositionRange().getStart());
Assert.assertEquals(lastest, entrys2.getPositionRange().getEnd());
Assert.assertEquals(first, CanalEventUtils.createPosition(entrys.get(0)));
Assert.assertEquals(lastest, CanalEventUtils.createPosition(entrys.get(entrys.size() - 1)));
// 全部ack掉
eventStore.cleanUntil(entrys2.getPositionRange().getEnd());
// 最后就拿不到数据
Events<Event> entrys3 = eventStore.tryGet(entrys1.getPositionRange().getEnd(), bufferSize);
System.out.println("third get size : " + entrys3.getEvents().size());
Assert.assertEquals(0, entrys3.getEvents().size());
eventStore.stop();
}
}