server模块的核心接口是CanalServer，其有2个实现类CanalServerWithNetty、CanalServerWithEmbeded。关于CanalServer，官方文档中有有以下描述：

  下图是笔者对官方文档的进一步描述：

左边的图

表示的是Canal独立部署。不同的应用通过canal client与canal server进行通信，所有的canal client的请求统一由CanalServerWithNetty接受，之后CanalServerWithNetty会将客户端请求派给CanalServerWithEmbeded 进行真正的处理。CannalServerWithEmbeded内部维护了多个canal instance，每个canal instance伪装成不同的mysql实例的slave，而CanalServerWithEmbeded会根据客户端请求携带的destination参数确定要由哪一个canal instance为其提供服务。

右边的图

是直接在应用中嵌入CanalServerWithEmbeded，不需要独立部署canal。很明显，网络通信环节少了，同步binlog信息的效率肯定更高。但是对于使用者的技术要求比较高。在应用中，我们可以通过CanalServerWithEmbeded.instance()方法来获得CanalServerWithEmbeded实例，这一个单例。

整个server模块源码目录结构如下所示：

其中上面的红色框就是嵌入式实现，而下面的绿色框是基于Netty的实现。

看起来基于netty的实现代码虽然多一点，这其实只是幻觉，CanalServerWithNetty会将所有的请求委派给CanalServerWithEmbedded处理。

而内嵌的方式只有CanalServerWithEmbedded一个类， 是因为CanalServerWithEmbedded又要根据destination选择某个具体的CanalInstance来处理客户端请求，而CanalInstance的实现位于instance模块，我们将在之后分析。因此从canal server的角度来说，CanalServerWithEmbedded才是server模块真正的核心。

        CanalServerWithNetty和CanalServerWithEmbedded都是单例的，提供了一个静态方法instance()获取对应的实例。回顾前一节分析CanalController源码时，在CanalController构造方法中准备CanalServer的相关代码，就是通过这两个静态方法获取对应的实例的。

1.  public CanalController(final Properties properties){
2.         ....
3.      // 准备canal server
4.         ip = getProperty(properties, CanalConstants.CANAL_IP);
5.         port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT));
6.         embededCanalServer = CanalServerWithEmbedded.instance();
7.         embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator
8.         canalServer = CanalServerWithNetty.instance();
9.         canalServer.setIp(ip);
10.         canalServer.setPort(port);
11.        ....   
12. }

CanalServer接口

 CanalServer接口继承了CanalLifeCycle接口，主要是为了重新定义start和stop方法，抛出CanalServerException。

1. public interface CanalServer extends CanalLifeCycle {
3.     void start() throws CanalServerException;
5.     void stop() throws CanalServerException;
6. }

CanalServerWithNetty

CanalServerWithNetty主要用于接受客户端的请求，然后将其委派给CanalServerWithEmbeded处理。下面的源码显示了CanalServerWithNetty种定义的字段和构造方法

1. public class CanalServerWithNetty extends AbstractCanalLifeCycle implements CanalServer {
2.     //监听的所有客户端请求都会为派给CanalServerWithEmbedded处理 
3.     private CanalServerWithEmbedded embeddedServer;      // 嵌入式server
5.     //监听的ip和port，client通过此ip和port与服务端通信
6.     private String                  ip;
7.     private int                     port;
9.     //netty组件
10.     private Channel                 serverChannel = null;
11.     private ServerBootstrap         bootstrap     = null;
12.     //....单例模式实现
13.     private CanalServerWithNetty(){
14.         //给embeddedServer赋值
15.         this.embeddedServer = CanalServerWithEmbedded.instance();
16.     }
17.     //... start and stop method
18.     //...setters and getters...
19. }

字段说明：

• embeddedServer：因为CanalServerWithNetty需要将请求委派给CanalServerWithEmbeded处理，因此其维护了embeddedServer对象。

• ip、port：这是netty监听的网络ip和端口，client通过这个ip和端口与server通信

• serverChannel、bootstrap：这是netty的API。其中ServerBootstrap用于启动服务端，通过调用其bind方法，返回一个类型为Channel的serverChannel对象，代表服务端通道。关于netty知识不是本教程重点，如果读者不熟悉，可以参考笔者的netty教程。

start方法

start方法中包含了netty启动的核心逻辑，如下所示：

com.alibaba.otter.canal.server.netty.CanalServerWithNetty#start

1. public void start() {
2.         super.start();
3.         //优先启动内嵌的canal server，因为基于netty的实现需要将请求委派给其处理
4.         if (!embeddedServer.isStart()) {
5.             embeddedServer.start();
6.         }
7.         
8.          /* 创建bootstrap实例，参数NioServerSocketChannelFactory也是Netty的API，其接受2个线程池参数
9.          其中第一个线程池是Accept线程池，第二个线程池是woker线程池，
10.          Accept线程池接收到client连接请求后，会将代表client的对象转发给worker线程池处理。
11.          这里属于netty的知识，不熟悉的用户暂时不必深究，简单认为netty使用线程来处理客户端的高并发请求即可。*/
12.         this.bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
13.             Executors.newCachedThreadPool()));
14.             
15.         /*pipeline实际上就是netty对客户端请求的处理器链，
16.         可以类比JAVA EE编程中Filter的责任链模式，上一个filter处理完成之后交给下一个filter处理，
17.         只不过在netty中，不再是filter，而是ChannelHandler。*/
18.         bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
19.             public ChannelPipeline getPipeline() throws Exception {
20.                 ChannelPipeline pipelines = Channels.pipeline();
21.                //主要是处理编码、解码。因为网路传输的传入的都是二进制流，FixedHeaderFrameDecoder的作用就是对其进行解析
22.                 pipelines.addLast(FixedHeaderFrameDecoder.class.getName(), new FixedHeaderFrameDecoder());
23.                //处理client与server握手
24.                 pipelines.addLast(HandshakeInitializationHandler.class.getName(), new HandshakeInitializationHandler());
25.                //client身份验证
26.                pipelines.addLast(ClientAuthenticationHandler.class.getName(),
27.                     new ClientAuthenticationHandler(embeddedServer));
28.                 //SessionHandler用于真正的处理客户端请求，是本文分析的重点
29.                SessionHandler sessionHandler = new SessionHandler(embeddedServer);
30.                 pipelines.addLast(SessionHandler.class.getName(), sessionHandler);
31.                 return pipelines;
32.             }
33.         });
34.         
35.         // 启动，当bind方法被调用时，netty开始真正的监控某个端口，此时客户端对这个端口的请求可以被接受到
36.         if (StringUtils.isNotEmpty(ip)) {
37.             this.serverChannel = bootstrap.bind(new InetSocketAddress(this.ip, this.port));
38.         } else {
39.             this.serverChannel = bootstrap.bind(new InetSocketAddress(this.port));
40.         }
41.     }

关于stop方法无非是一些关闭操作，代码很简单，这里不做介绍。

SessionHandler

    很明显的，canal处理client请求的核心逻辑都在SessionHandler这个处理器中。注意其在实例化时，传入了embeddedServer对象，前面我们提过，CanalServerWithNetty要将请求委派给CanalServerWithEmbedded处理，显然SessionHandler也要维护embeddedServer实例。

    这里我们主要分析SessionHandler的 messageReceived方法，这个方法表示接受到了一个客户端请求，我们主要看的是SessionHandler如何对客户端请求进行解析，然后委派给CanalServerWithEmbedded处理的。为了体现其转发请求处理的核心逻辑，以下代码省去了大量源码片段，如下

SessionHandler#messageReceived

1. public class SessionHandler extends SimpleChannelHandler {
2. ....
3. //messageReceived方法表示收到客户端请求
4. public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
5.     ....
6.       //根据客户端发送的网路通信包请求类型type，将请求委派embeddedServer处理
7.         switch (packet.getType()) {
8.             case SUBSCRIPTION://订阅请求
9.                 ...
10.                 embeddedServer.subscribe(clientIdentity);
11.                          ...
12.                 break;
13.             case UNSUBSCRIPTION://取消订阅请求
14.                 ...
15.                 embeddedServer.unsubscribe(clientIdentity);
16.                 ...
17.                 break;
18.             case GET://获取binlog请求
19.                 ....
20.                     if (get.getTimeout() == -1) {// 根据客户端是否指定了请求超时时间调用embeddedServer不同方法获取binlog
21.                         message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize());
22.                     } else {
23.                         ...
24.                         message = embeddedServer.getWithoutAck(clientIdentity,
25.                             get.getFetchSize(),
26.                             get.getTimeout(),
27.                             unit);
28.                     }
29.                 ...   
30.                 break;
31.             case CLIENTACK://客户端消费成功ack请求
32.                ...
33.                   embeddedServer.ack(clientIdentity, ack.getBatchId());
34.                ...
35.                 break;
36.             case CLIENTROLLBACK://客户端消费失败回滚请求
37.                 ...
38.                     if (rollback.getBatchId() == 0L) {
39.                         embeddedServer.rollback(clientIdentity);// 回滚所有批次
40.                     } else {
41.                         embeddedServer.rollback(clientIdentity, rollback.getBatchId()); // 只回滚单个批次
42.                     }
43.                 ...
44.                 break;
45.             default://无法判断请求类型
46.                 NettyUtils.error(400, MessageFormatter.format("packet type={} is NOT supported!", packet.getType())
47.                     .getMessage(), ctx.getChannel(), null);
48.                 break;
49.         }
50.     ...
51. }
52. ...
53. }

    可以看到，SessionHandler对client请求进行解析后，根据请求类型，委派给CanalServerWithEmbedded的相应方法进行处理。因此核心逻辑都在CanalServerWithEmbedded中。

CannalServerWithEmbeded

CanalServerWithEmbedded实现了CanalServer和CanalServiceCan两个接口。其内部维护了一个Map，key为destination，value为对应的CanalInstance，根据客户端请求携带的destination参数将其转发到对应的CanalInstance上去处理。

1. public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer, CanalService {
2.     ...
3.     //key为destination，value为对应的CanalInstance。
4.     private Map<String, CanalInstance> canalInstances;
5.     ...
6. }

对于CanalServer接口中定义的start和stop这两个方法实现比较简单，这里不再赘述。

     在上面的SessionHandler源码分析中，我们已经看到，会根据请求报文的类型，会调用CanalServerWithEmbedded的相应方法，这些方法都定义在CanalService接口中，如下：

1. public interface CanalService {
2.    //订阅
3.     void subscribe(ClientIdentity clientIdentity) throws CanalServerException;
4.    //取消订阅
5.     void unsubscribe(ClientIdentity clientIdentity) throws CanalServerException;
6.    //比例获取数据，并自动自行ack
7.     Message get(ClientIdentity clientIdentity, int batchSize) throws CanalServerException;
8.    //超时时间内批量获取数据，并自动进行ack
9.     Message get(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit) throws CanalServerException;
10.     //批量获取数据，不进行ack
11.     Message getWithoutAck(ClientIdentity clientIdentity, int batchSize) throws CanalServerException;
12.    //超时时间内批量获取数据，不进行ack
13.     Message getWithoutAck(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit)                                                                                               throws CanalServerException;
14.    //ack某个批次的数据
15.     void ack(ClientIdentity clientIdentity, long batchId) throws CanalServerException;
16.    //回滚所有没有ack的批次的数据
17.     void rollback(ClientIdentity clientIdentity) throws CanalServerException;
18.    //回滚某个批次的数据
19.     void rollback(ClientIdentity clientIdentity, Long batchId) throws CanalServerException;
20. }

 细心地的读者会发现，每个方法中都包含了一个ClientIdentity类型参数，这就是客户端身份的标识。

1. public class ClientIdentity implements Serializable {
2.     private String destination;
3.     private short  clientId;
4.     private String filter;
5.  ...
6. }

CanalServerWithEmbedded就是根据ClientIdentity中的destination参数确定这个请求要交给哪个CanalInstance处理的。

下面一次分析每一个方法的作用：

subscribe方法：

subscribe主要用于处理客户端的订阅请求，目前情况下，一个CanalInstance只能由一个客户端订阅，不过可以重复订阅。订阅主要的处理步骤如下：

1、根据客户端要订阅的destination，找到对应的CanalInstance

2、通过这个CanalInstance的CanalMetaManager组件记录下有客户端订阅。

3、获取客户端当前订阅位置(Position)。首先尝试从CanalMetaManager中获取，CanalMetaManager 中记录了某个client当前订阅binlog的位置信息。如果是第一次订阅，肯定无法获取到这个位置，则尝试从CanalEventStore中获取第一个binlog的位置。从CanalEventStore中获取binlog位置信息的逻辑是：CanalInstance一旦启动，就会立刻去拉取binlog，存储到CanalEventStore中，在第一次订阅的情况下，CanalEventStore中的第一条binlog的位置，就是当前客户端当前消费的开始位置。

4、通知CanalInstance订阅关系变化

1. /**
2.  * 客户端订阅，重复订阅时会更新对应的filter信息
3.  */
4. @Override
5. public void subscribe(ClientIdentity clientIdentity) throws CanalServerException {
6.     checkStart(clientIdentity.getDestination());
7.     //1、根据客户端要订阅的destination，找到对应的CanalInstance 
8.     CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
9.     if (!canalInstance.getMetaManager().isStart()) {
10.         canalInstance.getMetaManager().start();
11.     }
12.   //2、通过CanalInstance的CanalMetaManager组件进行元数据管理，记录一下当前这个CanalInstance有客户端在订阅
13.     canalInstance.getMetaManager().subscribe(clientIdentity); // 执行一下meta订阅
14.   //3、获取客户端当前订阅的binlog位置(Position)，首先尝试从CanalMetaManager中获取
15.     Position position = canalInstance.getMetaManager().getCursor(clientIdentity);
16.     if (position == null) {
17.   //3.1 如果是第一次订阅，尝试从CanalEventStore中获取第一个binlog的位置，作为客户端订阅开始的位置。
18.         position = canalInstance.getEventStore().getFirstPosition();// 获取一下store中的第一条
19.         if (position != null) {
20.             canalInstance.getMetaManager().updateCursor(clientIdentity, position); // 更新一下cursor
21.         }
22.         logger.info("subscribe successfully, {} with first position:{} ", clientIdentity, position);
23.     } else {
24.         logger.info("subscribe successfully, use last cursor position:{} ", clientIdentity, position);
25.     }
26.     //4 通知下订阅关系变化
27.     canalInstance.subscribeChange(clientIdentity);
28. }

unsubscribe方法：

   unsubscribe方法主要用于取消订阅关系。在下面的代码中，我们可以看到，其实就是找到CanalInstance对应的CanalMetaManager，调用其unsubscribe取消这个订阅记录。需要注意的是，取消订阅并不意味着停止CanalInstance。当某个客户端取消了订阅，还会有新的client来订阅这个CanalInstance，所以不能停。

1. /**
2.  * 取消订阅
3.  */
4. @Override
5. public void unsubscribe(ClientIdentity clientIdentity) throws CanalServerException {
6.     CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
7.     canalInstance.getMetaManager().unsubscribe(clientIdentity); // 执行一下meta订阅
8.     logger.info("unsubscribe successfully, {}", clientIdentity);
9. }

listAllSubscribe方法：

    这一个管理方法，其作用是列出订阅某个destination的所有client。这里返回的是一个List<ClientIdentity>，不过我们已经多次提到，目前一个destination只能由一个client订阅。这里之所以返回一个list，是canal原先计划要支持多个client订阅同一个destination。不过，这个功能一直没有实现。所以List中，实际上只会包含一个ClientIdentity。

1. /**
2.  * 查询所有的订阅信息
3.  */
4. public List<ClientIdentity> listAllSubscribe(String destination) throws CanalServerException {
5.     CanalInstance canalInstance = canalInstances.get(destination);
6.     return canalInstance.getMetaManager().listAllSubscribeInfo(destination);
7. }

listBatchIds方法:

1. /**
2.  * 查询当前未被ack的batch列表，batchId会按照从小到大进行返回
3.  */
4. public List<Long> listBatchIds(ClientIdentity clientIdentity) throws CanalServerException {
5.     checkStart(clientIdentity.getDestination());
6.     checkSubscribe(clientIdentity);
7.     CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
8.     Map<Long, PositionRange> batchs = canalInstance.getMetaManager().listAllBatchs(clientIdentity);
9.     List<Long> result = new ArrayList<Long>(batchs.keySet());
10.     Collections.sort(result);
11.     return result;
12. }

getWithoutAck方法：

getWithoutAck方法用于客户端获取binlog消息 ，一个获取一批(batch)的binlog，canal会为这批binlog生成一个唯一的batchId。客户端如果消费成功，则调用ack方法对这个批次进行确认。如果失败的话，可以调用rollback方法进行回滚。客户端可以连续多次调用getWithoutAck方法来获取binlog，在ack的时候，需要按照获取到binlog的先后顺序进行ack。如果后面获取的binlog被ack了，那么之前没有ack的binlog消息也会自动被ack。

getWithoutAck方法大致工作步骤如下所示：

1. 根据destination找到要从哪一个CanalInstance中获取binlog消息。

2. 确定从哪一个位置(Position)开始继续消费binlog。通常情况下，这个信息是存储在CanalMetaManager中。特别的，在第一次获取的时候，CanalMetaManager 中还没有存储任何binlog位置信息。此时CanalEventStore中存储的第一条binlog位置，则应该client开始消费的位置。

3. 根据Position从CanalEventStore中获取binlog。为了尽量提高效率，一般一次获取一批binlog，而不是获取一条。这个批次的大小(batchSize)由客户端指定。同时客户端可以指定超时时间，在超时时间内，如果获取到了batchSize的binlog，会立即返回。 如果超时了还没有获取到batchSize指定的binlog个数，也会立即返回。特别的，如果没有设置超时时间，如果没有获取到binlog也立即返回。

4. 在CanalMetaManager中记录这个批次的binlog消息。CanalMetaManager会为获取到的这个批次的binlog生成一个唯一的batchId，batchId是递增的。如果binlog信息为空，则直接把batchId设置为-1。 

5. @Override
6. public Message getWithoutAck(ClientIdentity clientIdentity, int batchSize) throws CanalServerException {
7.     return getWithoutAck(clientIdentity, batchSize, null, null);
8. }
9. /**
10.  * <pre>
11.  * 几种case:
12.  * a. 如果timeout为null，则采用tryGet方式，即时获取
13.  * b. 如果timeout不为null
14.  *    1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回
15.  *    2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少
16.  * </pre>
17.  */
18. @Override
19. public Message getWithoutAck(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit)
20.                                                                                                        throws CanalServerException {
21.     checkStart(clientIdentity.getDestination());
22.     checkSubscribe(clientIdentity);
23.       // 1、根据destination找到要从哪一个CanalInstance中获取binlog消息
24.     CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
25.     synchronized (canalInstance) {
26.         //2、从CanalMetaManager中获取最后一个没有ack的binlog批次的位置信息。
27.         PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().getLastestBatch(clientIdentity);
28.    //3 从CanalEventStore中获取binlog
29.         Events<Event> events = null;
30.         if (positionRanges != null) { // 3.1 如果从CanalMetaManager获取到了位置信息，从当前位置继续获取binlog
31.             events = getEvents(canalInstance.getEventStore(), positionRanges.getStart(), batchSize, timeout, unit);
32.         } else { //3.2 如果没有获取到binlog位置信息，从当前store中的第一条开始获取
33.             Position start = canalInstance.getMetaManager().getCursor(clientIdentity);
34.             if (start == null) { // 第一次，还没有过ack记录，则获取当前store中的第一条
35.                 start = canalInstance.getEventStore().getFirstPosition();
36.             }
37.       // 从CanalEventStore中获取binlog消息
38.             events = getEvents(canalInstance.getEventStore(), start, batchSize, timeout, unit);
39.         }
40.         //4 记录批次信息到CanalMetaManager中
41.         if (CollectionUtils.isEmpty(events.getEvents())) {
42.           //4.1 如果获取到的binlog消息为空，构造一个空的Message对象，将batchId设置为-1返回给客户端
43.             logger.debug("getWithoutAck successfully, clientId:{} batchSize:{} but result is null", new Object[] {
44.                     clientIdentity.getClientId(), batchSize });
45.             return new Message(-1, new ArrayList<Entry>()); // 返回空包，避免生成batchId，浪费性能
46.         } else {
47.            //4.2 如果获取到了binlog消息，将这个批次的binlog消息记录到CanalMetaMaager中，并生成一个唯一的batchId
48.             Long batchId = canalInstance.getMetaManager().addBatch(clientIdentity, events.getPositionRange());
49.             //将Events转为Entry
50.             List<Entry> entrys = Lists.transform(events.getEvents(), new Function<Event, Entry>() {
51.                 public Entry apply(Event input) {
52.                     return input.getEntry();
53.                 }
54.             });
55.             logger.info("getWithoutAck successfully, clientId:{} batchSize:{}  real size is {} and result is [batchId:{} , position:{}]",
56.                 clientIdentity.getClientId(),
57.                 batchSize,
58.                 entrys.size(),
59.                 batchId,
60.                 events.getPositionRange());
61.             //构造Message返回
62.             return new Message(batchId, entrys);
63.         }
64.     }
65. }
66. /**
67.  * 根据不同的参数，选择不同的方式获取数据
68.  */
69. private Events<Event> getEvents(CanalEventStore eventStore, Position start, int batchSize, Long timeout,
70.                                 TimeUnit unit) {
71.     if (timeout == null) {
72.         return eventStore.tryGet(start, batchSize);
73.     } else {
74.         try {
75.             if (timeout <= 0) {
76.                 return eventStore.get(start, batchSize);
77.             } else {
78.                 return eventStore.get(start, batchSize, timeout, unit);
79.             }
80.         } catch (Exception e) {
81.             throw new CanalServerException(e);
82.         }
83.     }
84. }

ack方法：

ack方法时客户端用户确认某个批次的binlog消费成功。进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。注意：进行反馈时必须按照batchId的顺序进行ack(需有客户端保证)

ack时需要做以下几件事情：

1. 从CanalMetaManager中，移除这个批次的信息。在getWithoutAck方法中，将批次的信息记录到了CanalMetaManager中，ack时移除。

2. 记录已经成功消费到的binlog位置，以便下一次获取的时候可以从这个位置开始，这是通过CanalMetaManager记录的。

3. 从CanalEventStore中，将这个批次的binlog内容移除。因为已经消费成功，继续保存这些已经消费过的binlog没有任何意义，只会白白占用内存。 

4. @Override
5. public void ack(ClientIdentity clientIdentity, long batchId) throws CanalServerException {
6.     checkStart(clientIdentity.getDestination());
7.     checkSubscribe(clientIdentity);
8.     CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
9.     PositionRange<LogPosition> positionRanges = null;
10.    //1 从CanalMetaManager中，移除这个批次的信息
11.     positionRanges = canalInstance.getMetaManager().removeBatch(clientIdentity, batchId); // 更新位置
12.     if (positionRanges == null) { // 说明是重复的ack/rollback
13.         throw new CanalServerException(String.format("ack error , clientId:%s batchId:%d is not exist , please check",
14.             clientIdentity.getClientId(),
15.             batchId));
16.     }
17.     //2、记录已经成功消费到的binlog位置，以便下一次获取的时候可以从这个位置开始，这是通过CanalMetaManager记录的
18.     if (positionRanges.getAck() != null) {
19.         canalInstance.getMetaManager().updateCursor(clientIdentity, positionRanges.getAck());
20.         logger.info("ack successfully, clientId:{} batchId:{} position:{}",
21.             clientIdentity.getClientId(),
22.             batchId,
23.             positionRanges);
24.     }
25.       //3、从CanalEventStore中，将这个批次的binlog内容移除
26.     canalInstance.getEventStore().ack(positionRanges.getEnd());
27. }

rollback方法：

1. /**
2.  * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿
3.  */
4. @Override
5. public void rollback(ClientIdentity clientIdentity) throws CanalServerException {
6.     checkStart(clientIdentity.getDestination());
7.     CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
8.     // 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
9.     boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
10.     if (!hasSubscribe) {
11.         return;
12.     }
13.     synchronized (canalInstance) {
14.         // 清除batch信息
15.         canalInstance.getMetaManager().clearAllBatchs(clientIdentity);
16.         // rollback eventStore中的状态信息
17.         canalInstance.getEventStore().rollback();
18.         logger.info("rollback successfully, clientId:{}", new Object[] { clientIdentity.getClientId() });
19.     }
20. }
21. /**
22.  * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿
23.  */
24. @Override
25. public void rollback(ClientIdentity clientIdentity, Long batchId) throws CanalServerException {
26.     checkStart(clientIdentity.getDestination());
27.     CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
28.     // 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
29.     boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
30.     if (!hasSubscribe) {
31.         return;
32.     }
33.     synchronized (canalInstance) {
34.         // 清除batch信息
35.         PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().removeBatch(clientIdentity,
36.             batchId);
37.         if (positionRanges == null) { // 说明是重复的ack/rollback
38.             throw new CanalServerException(String.format("rollback error, clientId:%s batchId:%d is not exist , please check",
39.                 clientIdentity.getClientId(),
40.                 batchId));
41.         }
42.         // lastRollbackPostions.put(clientIdentity,
43.         // positionRanges.getEnd());// 记录一下最后rollback的位置
44.         // TODO 后续rollback到指定的batchId位置
45.         canalInstance.getEventStore().rollback();// rollback
46.                                                  // eventStore中的状态信息
47.         logger.info("rollback successfully, clientId:{} batchId:{} position:{}",
48.             clientIdentity.getClientId(),
49.             batchId,
50.             positionRanges);
51.     }
52. }

get方法：

与getWithoutAck主要流程完全相同，唯一不同的是，在返回数据给用户前，直接进行了ack，而不管客户端消费是否成功。

1. @Override
2. public Message get(ClientIdentity clientIdentity, int batchSize) throws CanalServerException {
3.     return get(clientIdentity, batchSize, null, null);
4. }
5.  /*
6.  * 几种case:
7.  * a. 如果timeout为null，则采用tryGet方式，即时获取
8.  * b. 如果timeout不为null
9.  *    1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回
10.  *    2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少
11.  */
12. @Override
13. public Message get(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit)
14.                                                                                              throws CanalServerException {
15.     checkStart(clientIdentity.getDestination());
16.     checkSubscribe(clientIdentity);
17.     CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
18.     synchronized (canalInstance) {
19.         // 获取到流式数据中的最后一批获取的位置
20.         PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().getLastestBatch(clientIdentity);
21.         if (positionRanges != null) {
22.             throw new CanalServerException(String.format("clientId:%s has last batch:[%s] isn't ack , maybe loss data",
23.                 clientIdentity.getClientId(),
24.                 positionRanges));
25.         }
26.         Events<Event> events = null;
27.         Position start = canalInstance.getMetaManager().getCursor(clientIdentity);
28.         events = getEvents(canalInstance.getEventStore(), start, batchSize, timeout, unit);
29.         if (CollectionUtils.isEmpty(events.getEvents())) {
30.             logger.debug("get successfully, clientId:{} batchSize:{} but result is null", new Object[] {
31.                     clientIdentity.getClientId(), batchSize });
32.             return new Message(-1, new ArrayList<Entry>()); // 返回空包，避免生成batchId，浪费性能
33.         } else {
34.             // 记录到流式信息
35.             Long batchId = canalInstance.getMetaManager().addBatch(clientIdentity, events.getPositionRange());
36.             List<Entry> entrys = Lists.transform(events.getEvents(), new Function<Event, Entry>() {
37.                 public Entry apply(Event input) {
38.                     return input.getEntry();
39.                 }
40.             });
41.             logger.info("get successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]",
42.                 clientIdentity.getClientId(),
43.                 batchSize,
44.                 entrys.size(),
45.                 batchId,
46.                 events.getPositionRange());
47.             // 直接提交ack
48.             ack(clientIdentity, batchId);
49.             return new Message(batchId, entrys);
50.         }
51.     }
52. }