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
