【Canal源码分析】Sink及Store工作过程

一、序列图

二、源码分析

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 10msLockSupport.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 emptynotEmpty.signal();
}

这里主要对put一些指针，还有空间做了重新的计算。放到队列中之后，通知其他等待notEmpty的线程，来进行数据的获取，这时候，client可以进行数据获取了。