一、序列图

二、源码分析

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可以进行数据获取了。