Canal源码分析----MetaManager

MetaManager即meta信息管理器，在Canal中主要用来管理一些元数据的信息，下面以内存版实现MemoryMetaManager为例来分析

public class MemoryMetaManager extends AbstractCanalLifeCycle implements CanalMetaManager {

    protected Map<String, List<ClientIdentity>>              destinations;
    protected Map<ClientIdentity, MemoryClientIdentityBatch> batches;
    protected Map<ClientIdentity, Position>                  cursors;

    public static class MemoryClientIdentityBatch {

        private ClientIdentity           clientIdentity;
        private Map<Long, PositionRange> batches          = new MapMaker().makeMap();
        private AtomicLong               atomicMaxBatchId = new AtomicLong(1);

    }
}
public class ClientIdentity implements Serializable {

    private String destination;//instance的name
    private short  clientId;//client的Id
    private String filter;//Client指定的filter

}
public class PositionRange<T extends Position> implements Serializable {

    private static final long serialVersionUID = -9162037079815694784L;
    private T                 start;
    // add by ljh at 2012-09-05，用于记录一个可被ack的位置，保证每次提交到cursor中的位置是一个完整事务的结束
    private T                 ack;
    private T                 end;

}

destinations：保存着Client端信息，key为ClientId
batches：保存着Client对应的未ack的记录，key为ClientId
cursors：保存着Client对应的ack的位置，key为ClientId
MemoryClientIdentityBatch：

batches：未ack的位置范围，key为batchId(每次拉取数据都会给客户端返回一个batchId)
atomicMaxBatchId：最大的batchId，递增

上面就是MetaManager的数据结构，那么MetaManager有什么功能了，先看下接口主要的方法声明

public interface CanalMetaManager extends CanalLifeCycle {

    void subscribe(ClientIdentity clientIdentity) throws CanalMetaManagerException;

    void unsubscribe(ClientIdentity clientIdentity) throws CanalMetaManagerException;

    Position getCursor(ClientIdentity clientIdentity) throws CanalMetaManagerException;

    void updateCursor(ClientIdentity clientIdentity, Position position) throws CanalMetaManagerException;

    PositionRange getFirstBatch(ClientIdentity clientIdentity) throws CanalMetaManagerException;

    PositionRange getLastestBatch(ClientIdentity clientIdentity) throws CanalMetaManagerException;

    Long addBatch(ClientIdentity clientIdentity, PositionRange positionRange) throws CanalMetaManagerException;

    void addBatch(ClientIdentity clientIdentity, PositionRange positionRange, Long batchId)
                                                                                           throws CanalMetaManagerException;
    PositionRange getBatch(ClientIdentity clientIdentity, Long batchId) throws CanalMetaManagerException;

    PositionRange removeBatch(ClientIdentity clientIdentity, Long batchId) throws CanalMetaManagerException;

    void clearAllBatchs(ClientIdentity clientIdentity) throws CanalMetaManagerException;

}

subscribe：以destination做key，保存对应的ClientIdentity信息到destinations中
unsubscribe：从destinations中移除对于的Client信息
以上两个方法在当Client发起subscribe/unsubscribe请求的时候会调用

getCursor：获取Client Ack到的位置信息
例如在Server处理Client的get请求的时候，会先获取该信息，来判断应该获取多少数据

updateCursor：更新Client Ack到的位置信息
例如在Server处理Client的ack请求的时候，会更新该信息

getFirstBatch：获取第一个未确认的位置信息，即从MemoryClientIdentityBatch的batches中获取batchId最小的位置信息。因为batchId是递增的，batchId最小的即为第一个
getLastestBatch：与getFirstBatch相反。
例如在Server处理Client的get请求的时候，会先获取未确认的信息进行处理

addBatch：将一个未确认的位置信息添加到MetaManager
例如在Server处理Client的get请求的时候，会取到当次获取的数据的起始位置，即为PositionRange，将其添加进MetaManager

removeBatch：确认或者回滚后将位置信息从MetaManager中移除

clearAllBatchs：回滚后将位置信息从MetaManager中移除

上面就是内存版MetaManager的功能，MetaManager有多个实现，
基于ZK：结构和内存实现类似
基于内存：即上面说的
基于混合模式：实现和内存的类似，但是会定时将数据上传到zk

Canal源码分析----MetaManager

推荐阅读更多精彩内容