一、序列图
二、源码分析
2.1 Sink
Sink阶段所做的事情，就是根据一定的规则，对binlog数据进行一定的过滤。我们之前跟踪过parser过程的代码，发现在parser完成后，会把数据放到一个环形队列TransactionBuffer中，也就是这个方法：
transactionBuffer.add(entry);
我们具体看下add这个方法。
public void add(CanalEntry.Entry entry) throws InterruptedException {
switch (entry.getEntryType()) {
case TRANSACTIONBEGIN:
flush();// 刷新上一次的数据
put(entry);
break;
case TRANSACTIONEND:
put(entry);
flush();
break;
case ROWDATA:
put(entry);
// 针对非DML的数据，直接输出，不进行buffer控制
EventType eventType = entry.getHeader().getEventType();
if (eventType != null && !isDml(eventType)) {
flush();
}
break;
default:
break;
}
}
判断一下事件的类型，如果是事务开头，那么直接刷新之前的数据，然后把当前事件加到队列中；如果是事务的结束，那么先把当前事务放到队列后，刷新到下一个阶段；如果是普通的事件，直接放到队列中，如果事务头类型不为空，且不是DML类型，那么直接刷新队列中数据到下一个阶段。
我们需要理清楚这块的逻辑，什么时候flush，什么时候put，针对不同的事件，采取的策略不一样。
这里我们分析下flush和put两个步骤。
2.1.1 flush队列
这块其实还没有涉及到sink阶段，还在维护一个事件环形队列。这个环形队列，维护了两个指针，一个是flush的指针，一个是put的指针，flush的指针永远是滞后于put指针的。
private void flush() throws InterruptedException {
long start = this.flushSequence.get() + 1;
long end = this.putSequence.get();
if (start <= end) {
List<CanalEntry.Entry> transaction = new ArrayList<CanalEntry.Entry>();
for (long next = start; next <= end; next++) {
transaction.add(this.entries[getIndex(next)]);
}
flushCallback.flush(transaction);
flushSequence.set(end);// flush成功后，更新flush位置
}
}
start就是flush的指针，end就是put的指针，flush的动作就是把当前flush到put中间的数据，全部刷新到下一个阶段。具体传递到下一个阶段的代码在flushCallback.flush方法中。这块我们下文再分析。
2.1.2 put
private void put(CanalEntry.Entry data) throws InterruptedException {
// 首先检查是否有空位
if (checkFreeSlotAt(putSequence.get() + 1)) {
long current = putSequence.get();
long next = current + 1;
// 先写数据，再更新对应的cursor,并发度高的情况，putSequence会被get请求可见，拿出了ringbuffer中的老的Entry值
entries[getIndex(next)] = data;
putSequence.set(next);
} else {
flush();// buffer区满了，刷新一下
put(data);// 继续加一下新数据
}
}
这块的注释都比较清晰了，就不赘述了。
2.1.3 flush到sink
具体的代码在AbstractEventParser中，定义transactionBuffer的地方。
public void flush(List<CanalEntry.Entry> transaction) throws InterruptedException {
boolean successed = consumeTheEventAndProfilingIfNecessary(transaction);
if (!running) {
return;
}
if (!successed) {
throw new CanalParseException("consume failed!");
}
LogPosition position = buildLastTransactionPosition(transaction);
if (position != null) { // 可能position为空
logPositionManager.persistLogPosition(AbstractEventParser.this.destination, position);
}
}
主要的处理在consumeTheEventAndProfilingIfNecessary里面。这里面调用了eventSink.sink()方法。
2.1.4 sink
这里面进行了binlog数据的过滤。首先判断是否需要过滤事务头和尾，如果需要过滤的话，直接过滤掉，默认不过滤。
遍历传到这个阶段的binlog列表，根据正则表达式判断，是否需要进行过滤，一般来说是根据表名、库名等进行过滤。这边的过滤类主要是AviaterRegexFilter，根据库名.表名和表达式进行过滤。如果需要进行过滤，那么直接把这个事件过滤。否则，加到binlog列表中，进行二次过滤。第二次过滤的主要内容是HEARTBEAT类型的事件，主要的代码在这里：
protected boolean doSink(List<Event> events) {
for (CanalEventDownStreamHandler<List<Event>> handler : getHandlers()) {
events = handler.before(events);//处理heartbeat事件
}
int fullTimes = 0;
do {
if (eventStore.tryPut(events)) {
for (CanalEventDownStreamHandler<List<Event>> handler : getHandlers()) {
events = handler.after(events);
}
return true;
} else {
applyWait(++fullTimes);
}
for (CanalEventDownStreamHandler<List<Event>> handler : getHandlers()) {
events = handler.retry(events);
}
} while (running && !Thread.interrupted());
return false;
}
这里的CanalEventDownStreamHandler其实只有HeartBeatEntryEventHandler，也就是在before方法中把heartbeat事件从events去掉。这个心跳事件其实是parser过程生成的，我们之前有提到过。after目前是空的方法。
去掉之后，剩余的事件列表就会被调用tryPut()方法，送到下一步骤store中。
这里还有个applyWait方法，防止无限等待。
private void applyWait(int fullTimes) {
int newFullTimes = fullTimes > maxFullTimes ? maxFullTimes : fullTimes;
if (fullTimes <= 3) { // 3次以内
Thread.yield();
} else { // 超过3次，最多只sleep 10ms
LockSupport.parkNanos(1000 * 1000L * newFullTimes);
}
}
2.2 Store
目前只有基于内存模式的Store，这个阶段是真正Server中的落盘阶段。数据经历了mysql master到parser，再到sink，最后终于到了这里。
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();
}
}
在进行数据put的时候，加了一把锁。首先计算下是否还有剩余的空间进行数据处理，这里的计算，不光是计算了队列的剩余长度，还计算了剩余空间。队列的长度默认是16*1024，如果空间不足，直接拒绝，返回false，等待空间空余出来后，再进行put操作。否则，直接doPut()。
/**
* 执行具体的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();
}
这里主要对put一些指针，还有空间做了重新的计算。放到队列中之后，通知其他等待notEmpty的线程，来进行数据的获取，这时候，client可以进行数据获取了。
