万字带你深入阿里开源的 Canal 工作原理

public class CanalServerWithNetty extends AbstractCanalLifeCycle implements CanalServer {    public void start() {        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {            public ChannelPipeline getPipeline() throws Exception {                ChannelPipeline pipelines = Channels.pipeline();                pipelines.addLast(FixedHeaderFrameDecoder.class.getName(), new FixedHeaderFrameDecoder());                // 处理客户端的HANDSHAKE请求                pipelines.addLast(HandshakeInitializationHandler.class.getName(),                    new HandshakeInitializationHandler(childGroups));                // 处理客户端的CLIENTAUTHENTICATION请求                pipelines.addLast(ClientAuthenticationHandler.class.getName(),                    new ClientAuthenticationHandler(embeddedServer));
                // 处理客户端的会话请求，包括SUBSCRIPTION，GET等                SessionHandler sessionHandler = new SessionHandler(embeddedServer);                pipelines.addLast(SessionHandler.class.getName(), sessionHandler);                return pipelines;            }        });    }}

public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {    waitClientRunning();    int size = (batchSize <= 0) ? 1000 : batchSize;    long time = (timeout == null || timeout < 0) ? -1 : timeout; // -1代表不做timeout控制    if (unit == null) unit = TimeUnit.MILLISECONDS;  //默认是毫秒
    // client发送GET请求    writeWithHeader(Packet.newBuilder()        .setType(PacketType.GET)        .setBody(Get.newBuilder()            .setAutoAck(false)            .setDestination(clientIdentity.getDestination())            .setClientId(String.valueOf(clientIdentity.getClientId()))            .setFetchSize(size)            .setTimeout(time)            .setUnit(unit.ordinal())            .build()            .toByteString())        .build()        .toByteArray());    // client获取GET结果        return receiveMessages();}
private Message receiveMessages() throws IOException {    // 读取server发送的数据包    Packet p = Packet.parseFrom(readNextPacket());    switch (p.getType()) {        case MESSAGES: {            Messages messages = Messages.parseFrom(p.getBody());            Message result = new Message(messages.getBatchId());            for (ByteString byteString : messages.getMessagesList()) {                result.addEntry(Entry.parseFrom(byteString));            }            return result;        }    }}

case GET:    // 读取客户端发送的数据包，封装为Get对象    Get get = CanalPacket.Get.parseFrom(packet.getBody());    // destination表示canal instance    if (StringUtils.isNotEmpty(get.getDestination()) && StringUtils.isNotEmpty(get.getClientId())) {        clientIdentity = new ClientIdentity(get.getDestination(), Short.valueOf(get.getClientId()));        Message message = null;        if (get.getTimeout() == -1) {// 是否是初始值            message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize());        } else {            TimeUnit unit = convertTimeUnit(get.getUnit());            message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize(), get.getTimeout(), unit);        }        // 设置返回给客户端的数据包类型为MESSAGES           Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder();        packetBuilder.setType(PacketType.MESSAGES);        // 构造Message        Messages.Builder messageBuilder = CanalPacket.Messages.newBuilder();        messageBuilder.setBatchId(message.getId());        if (message.getId() != -1 && !CollectionUtils.isEmpty(message.getEntries())) {            for (Entry entry : message.getEntries()) {                messageBuilder.addMessages(entry.toByteString());            }        }        packetBuilder.setBody(messageBuilder.build().toByteString());        // 输出数据，返回给客户端        NettyUtils.write(ctx.getChannel(), packetBuilder.build().toByteArray(), null);    }

Entry      Header          logfileName [binlog文件名]          logfileOffset [binlog position]          executeTime [binlog里记录变更发生的时间戳,精确到秒]          schemaName           tableName          eventType [insert/update/delete类型]      entryType   [事务头BEGIN/事务尾END/数据ROWDATA]      storeValue  [byte数据,可展开，对应的类型为RowChange]        RowChange      isDdl       [是否是ddl变更操作，比如create table/drop table]      sql         [具体的ddl sql]      rowDatas    [具体insert/update/delete的变更数据，可为多条，1个binlog event事件可对应多条变更，比如批处理]          beforeColumns [Column类型的数组，变更前的数据字段]          afterColumns [Column类型的数组，变更后的数据字段]            Column       index             sqlType     [jdbc type]      name        [column name]      isKey       [是否为主键]      updated     [是否发生过变更]      isNull      [值是否为null]      value       [具体的内容，注意为string文本]

case SUBSCRIPTION:        Sub sub = Sub.parseFrom(packet.getBody());        embeddedServer.subscribe(clientIdentity);case GET:        Get get = CanalPacket.Get.parseFrom(packet.getBody());        message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize());case CLIENTACK:        ClientAck ack = CanalPacket.ClientAck.parseFrom(packet.getBody());        embeddedServer.ack(clientIdentity, ack.getBatchId());case CLIENTROLLBACK:        ClientRollback rollback = CanalPacket.ClientRollback.parseFrom(packet.getBody());        embeddedServer.rollback(clientIdentity);// 回滚所有批次

public void subscribe(ClientIdentity clientIdentity) throws CanalServerException {
    // ClientIdentity表示Canal Client客户端，从中可以获取出客户端指定连接的Destination
    // 由于CanalServerWithEmbedded记录了每个Destination对应的Instance，可以获取客户端对应的Instance
    CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
    if (!canalInstance.getMetaManager().isStart()) {
        canalInstance.getMetaManager().start(); // 启动Instance的元数据管理器
    }
    canalInstance.getMetaManager().subscribe(clientIdentity); // 执行一下meta订阅
    Position position = canalInstance.getMetaManager().getCursor(clientIdentity);
    if (position == null) {
        position = canalInstance.getEventStore().getFirstPosition();// 获取一下store中的第一条
        if (position != null) {
            canalInstance.getMetaManager().updateCursor(clientIdentity, position); // 更新一下cursor
        }
    }
    // 通知下订阅关系变化
    canalInstance.subscribeChange(clientIdentity);
}

每个CanalInstance中包括了四个组件：**EventParser、EventSink、EventStore、MetaManager**。
服务端主要的处理方法包括get/ack/rollback，这三个方法都会用到Instance上面的几个内部组件，主要还是EventStore和MetaManager：
在这之前，要先理解EventStore的含义，EventStore是一个RingBuffer，有三个指针：**Put、Get、Ack**。
- Put: Canal Server从MySQL拉取到数据后，放到内存中，Put增加- Get: 消费者（Canal Client）从内存中消费数据，Get增加- Ack: 消费者消费完成，Ack增加。并且会删除Put中已经被Ack的数据
这三个操作与Instance组件的关系如下：![](https://img-blog.csdnimg.cn/20210120154723227.png? )客户端通过canal server获取mysql binlog有几种方式（get方法和getWithoutAck）：
- 如果timeout为null，则采用tryGet方式，即时获取- 如果timeout不为null  1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回  2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少```javaprivate Events<Event> getEvents(CanalEventStore eventStore, Position start, int batchSize, Long timeout,                                TimeUnit unit) {    if (timeout == null) {        return eventStore.tryGet(start, batchSize); // 即时获取    } else if (timeout <= 0){        return eventStore.get(start, batchSize); // 阻塞获取    } else {        return eventStore.get(start, batchSize, timeout, unit); // 异步获取    }}

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操作的最后一条的位置
    // 启动EventStore时，创建指定大小的缓冲区，Event数组的大小是16*1024    // 也就是说算个数的话，数组可以容纳16000个事件。算内存的话，大小为16MB    public void start() throws CanalStoreException {        super.start();        indexMask = bufferSize - 1;        entries = new Event[bufferSize];    }
    // EventParser解析后，会放入内存中（Event数组，缓冲区）    private void doPut(List<Event> data) {        long current = putSequence.get(); // 取得当前的位置，初始时为-1，第一个元素为-1+1=0        long end = current + data.size(); // 最末尾的位置，假设Put了10条数据，end=-1+10=9        // 先写数据，再更新对应的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);    } }

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;    }
    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);        }    }    // 处理PositionRange，然后设置getSequence为end    getSequence.compareAndSet(current, end)}

public void ack(Position position) throws CanalStoreException {    cleanUntil(position);}
public void cleanUntil(Position position) throws CanalStoreException {    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                }            }
            ackSequence.compareAndSet(sequence, next)        }    }}

public void rollback() throws CanalStoreException {    getSequence.set(ackSequence.get());    getMemSize.set(ackMemSize.get());}