client的工作过程，需要我们自己去编写对应的逻辑，我们目前只能从example写的例子来看。目前examle中提供了两个例子，一个是单机的，一个是集群的cluster，我们后续如果需要进行开发的话，其实也是开发我们自己的client，以及client的一些逻辑。我们主要看下集群的client是如何实现和消费的，又是怎么和server进行数据交互的。

我们来看看具体的代码：

protected void process() {
int batchSize = 5 * 1024;
while (running) {
try {
MDC.put("destination", destination);
connector.connect();
connector.subscribe();
waiting = false;
while (running) {
Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
long batchId = message.getId();
int size = message.getEntries().size();
if (batchId == -1 || size == 0) {
// try {
// Thread.sleep(1000);
// } catch (InterruptedException e) {
// }
} else {
printSummary(message, batchId, size);
printEntry(message.getEntries());
}
connector.ack(batchId); // 提交确认
// connector.rollback(batchId); // 处理失败, 回滚数据
}
} catch (Exception e) {
logger.error("process error!", e);
} finally {
connector.disconnect();
MDC.remove("destination");
}
}
}

这个的这样的过程是这样的

• 连接，connector.connect()
• 订阅，connector.subscribe
• 获取数据，connector.getWithoutAck()
• 业务处理
• 提交确认,connector.ack()
• 回滚，connector.rollback()
• 断开连接，connector.disconnect()

我们具体来看下。

一、建立连接

CanalConnector主要有两个实现，一个是SimpleCanalConnector，一个是ClusterCanalConnector，我们主要看下ClusterCanalConnector，这也是我们要用的一个模式。

我们用的时候，通过一个工厂类生成我们需要的Connector，这里的工厂类是CanalConnectors，里面包含了生成ClusterCanalConnector的方法。

public static CanalConnector newClusterConnector(String zkServers, String destination, String username,
String password) {
ClusterCanalConnector canalConnector = new ClusterCanalConnector(username,
password,
destination,
new ClusterNodeAccessStrategy(destination, ZkClientx.getZkClient(zkServers)));
canalConnector.setSoTimeout(30 * 1000);
return canalConnector;
}

用到的参数有zk的地址，canal的名称，数据库的账号密码。里面有个ClusterNodeAccessStrategy是用来选择client的策略，这个ClusterNodeAccessStrategy的构造方法里面有些东西需要我们关注下。

1.1 ClusterNodeAccessStrategy

public ClusterNodeAccessStrategy(String destination, ZkClientx zkClient){
this.zkClient = zkClient;
childListener = new IZkChildListener() {
public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
initClusters(currentChilds);
}
};
dataListener = new IZkDataListener() {
public void handleDataDeleted(String dataPath) throws Exception {
runningAddress = null;
}
public void handleDataChange(String dataPath, Object data) throws Exception {
initRunning(data);
}
};
String clusterPath = ZookeeperPathUtils.getDestinationClusterRoot(destination);
this.zkClient.subscribeChildChanges(clusterPath, childListener);
initClusters(this.zkClient.getChildren(clusterPath));
String runningPath = ZookeeperPathUtils.getDestinationServerRunning(destination);
this.zkClient.subscribeDataChanges(runningPath, dataListener);
initRunning(this.zkClient.readData(runningPath, true));
}

这边起了两个监听器，都是监听server端的活动服务器的。一个是获取所有的server列表，一个是获取活动的server服务器，都是从zk的对应节点上去取的。

1.2 连接connect

获取到CanalConnector之后，就是真正的连接了。在ClusterCanalConnector中，我们可以看到，其实他底层用的也是SimpleCanalConnector，只不过加了一个选择的策略。

public void connect() throws CanalClientException {
if (connected) {
return;
}
if (runningMonitor != null) {
if (!runningMonitor.isStart()) {
runningMonitor.start();
}
} else {
waitClientRunning();
if (!running) {
return;
}
doConnect();
if (filter != null) { // 如果存在条件，说明是自动切换，基于上一次的条件订阅一次
subscribe(filter);
}
if (rollbackOnConnect) {
rollback();
}
}
connected = true;
}

如果是集群模式的客户端，那么这边的runningMonitor不为空，因为他进行了初始化。我们主要看下runningMonitor.start()里面的操作。

public void start() {
super.start();
String path = ZookeeperPathUtils.getDestinationClientRunning(this.destination, clientData.getClientId());
zkClient.subscribeDataChanges(path, dataListener);
initRunning();
}

这边监听的路径是：/otter/canal/destinations/{destination}/{clientId}/running。如果有任何的变化，或节点的删除，那么执行dataListener里面的操作。

dataListener = new IZkDataListener() {
public void handleDataChange(String dataPath, Object data) throws Exception {
MDC.put("destination", destination);
ClientRunningData runningData = JsonUtils.unmarshalFromByte((byte[]) data, ClientRunningData.class);
if (!isMine(runningData.getAddress())) {
mutex.set(false);
}
if (!runningData.isActive() && isMine(runningData.getAddress())) { // 说明出现了主动释放的操作，并且本机之前是active
release = true;
releaseRunning();// 彻底释放mainstem
}
activeData = (ClientRunningData) runningData;
}
public void handleDataDeleted(String dataPath) throws Exception {
MDC.put("destination", destination);
mutex.set(false);
// 触发一下退出,可能是人为干预的释放操作或者网络闪断引起的session expired timeout
processActiveExit();
if (!release && activeData != null && isMine(activeData.getAddress())) {
// 如果上一次active的状态就是本机，则即时触发一下active抢占
initRunning();
} else {
// 否则就是等待delayTime，避免因网络瞬端或者zk异常，导致出现频繁的切换操作
delayExector.schedule(new Runnable() {
public void run() {
initRunning();
}
}, delayTime, TimeUnit.SECONDS);
}
}
};

这里的注释比较清楚，基本上如果数据发生了变化，那么进行节点释放后，将运行节点置为活动节点。如果发生了数据删除，那么直接触发退出，如果上一次的active状态是本机，那么触发一下active抢占，否则等待delayTime，默认5s后重试。下面我们主要看下initRunning。

1.3 initRunning

这块主要是创建运行节点的临时节点。节点路径是/otter/canal/destinations/{destination}/{clientId}，节点内容是ClientRunningData的json序列化结果。连接的代码：

public InetSocketAddress processActiveEnter() {
InetSocketAddress address = doConnect();
mutex.set(true);
if (filter != null) { // 如果存在条件，说明是自动切换，基于上一次的条件订阅一次
subscribe(filter);
}
if (rollbackOnConnect) {
rollback();
}
return address;
}

这块有几段逻辑，我们慢慢看下。

1.3.1 doConnect

这里是client直接连上了server，通过socket连接，也就是server暴露的socket端口。

private InetSocketAddress doConnect() throws CanalClientException {
try {
channel = SocketChannel.open();
channel.socket().setSoTimeout(soTimeout);
SocketAddress address = getAddress();
if (address == null) {
address = getNextAddress();
}
channel.connect(address);
readableChannel = Channels.newChannel(channel.socket().getInputStream());
writableChannel = Channels.newChannel(channel.socket().getOutputStream());
Packet p = Packet.parseFrom(readNextPacket());
if (p.getVersion() != 1) {
throw new CanalClientException("unsupported version at this client.");
}
if (p.getType() != PacketType.HANDSHAKE) {
throw new CanalClientException("expect handshake but found other type.");
}
//
Handshake handshake = Handshake.parseFrom(p.getBody());
supportedCompressions.addAll(handshake.getSupportedCompressionsList());
//
ClientAuth ca = ClientAuth.newBuilder()
.setUsername(username != null ? username : "")
.setPassword(ByteString.copyFromUtf8(password != null ? password : ""))
.setNetReadTimeout(soTimeout)
.setNetWriteTimeout(soTimeout)
.build();
writeWithHeader(Packet.newBuilder()
.setType(PacketType.CLIENTAUTHENTICATION)
.setBody(ca.toByteString())
.build()
.toByteArray());
//
Packet ack = Packet.parseFrom(readNextPacket());
if (ack.getType() != PacketType.ACK) {
throw new CanalClientException("unexpected packet type when ack is expected");
}
Ack ackBody = Ack.parseFrom(ack.getBody());
if (ackBody.getErrorCode() > 0) {
throw new CanalClientException("something goes wrong when doing authentication: "
+ ackBody.getErrorMessage());
}
connected = true;
return new InetSocketAddress(channel.socket().getLocalAddress(), channel.socket().getLocalPort());
} catch (IOException e) {
throw new CanalClientException(e);
}
}

这边采用NIO编程，建立和server的socket连接后，发送了握手包和认证包，当收到ack包后，认为连接成功。认证包的服务端处理在ClientAuthenticationHandler类中，握手处理在HandshakeInitializationHandler类。

server接收到认证的消息后，会做如下的处理：

public void messageReceived(final ChannelHandlerContext ctx, MessageEvent e) throws Exception {
ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
final Packet packet = Packet.parseFrom(buffer.readBytes(buffer.readableBytes()).array());
switch (packet.getVersion()) {
case SUPPORTED_VERSION:
default:
final ClientAuth clientAuth = ClientAuth.parseFrom(packet.getBody());
// 如果存在订阅信息
if (StringUtils.isNotEmpty(clientAuth.getDestination())
&& StringUtils.isNotEmpty(clientAuth.getClientId())) {
ClientIdentity clientIdentity = new ClientIdentity(clientAuth.getDestination(),
Short.valueOf(clientAuth.getClientId()),
clientAuth.getFilter());
try {
MDC.put("destination", clientIdentity.getDestination());
embeddedServer.subscribe(clientIdentity);
ctx.setAttachment(clientIdentity);// 设置状态数据
// 尝试启动，如果已经启动，忽略
if (!embeddedServer.isStart(clientIdentity.getDestination())) {
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination());
if (!runningMonitor.isStart()) {
runningMonitor.start();
}
}
} finally {
MDC.remove("destination");
}
}
NettyUtils.ack(ctx.getChannel(), new ChannelFutureListener() {
public void operationComplete(ChannelFuture future) throws Exception {
//忽略
}
});
break;
}
}

主要的逻辑在subscribe里面。如果metaManager没有启动，那么需要进行启动。启动时，会从zk节点下面拉取一些数据，包括客户端的消费位点情况等等。然后就是订阅，订阅是新建一个zk节点，路径为/otter/canal/destinations/{destination}/{clientId}。然后还有一些过滤器，也需要写到zk中。之后就是获取一下本client的位点信息，如果原来zk中包含，那么直接从内存中获取，否则取eventStore的第一条数据。

1.3.2 subscribe

发送订阅消息给server，通过socket的方式。这边是判断，如果filter不为空，才发送订阅消息。服务端的处理过程是这样的：

case SUBSCRIPTION:
Sub sub = Sub.parseFrom(packet.getBody());
if (StringUtils.isNotEmpty(sub.getDestination()) && StringUtils.isNotEmpty(sub.getClientId())) {
clientIdentity = new ClientIdentity(sub.getDestination(),
Short.valueOf(sub.getClientId()),
sub.getFilter());
MDC.put("destination", clientIdentity.getDestination());
// 尝试启动，如果已经启动，忽略
if (!embeddedServer.isStart(clientIdentity.getDestination())) {
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination());
if (!runningMonitor.isStart()) {
runningMonitor.start();
}
}
embeddedServer.subscribe(clientIdentity);
ctx.setAttachment(clientIdentity);// 设置状态数据
NettyUtils.ack(ctx.getChannel(), null);
} else {
NettyUtils.error(401,
MessageFormatter.format("destination or clientId is null", sub.toString()).getMessage(),
ctx.getChannel(),
null);
}
break;

类似于connect的过程，不过这边带上了filter的参数。这边启动了server以及他的监听器。

1.3.3 rollback

这里的回滚是指回滚server端记录的本client的位点信息。

public void rollback() throws CanalClientException {
waitClientRunning();
rollback(0);// 0代笔未设置
}

这里发送了rollback的指令。服务端是这么处理的：

case CLIENTROLLBACK:
ClientRollback rollback = CanalPacket.ClientRollback.parseFrom(packet.getBody());
MDC.put("destination", rollback.getDestination());
if (StringUtils.isNotEmpty(rollback.getDestination())
&& StringUtils.isNotEmpty(rollback.getClientId())) {
clientIdentity = new ClientIdentity(rollback.getDestination(),
Short.valueOf(rollback.getClientId()));
if (rollback.getBatchId() == 0L) {
embeddedServer.rollback(clientIdentity);// 回滚所有批次
} else {
embeddedServer.rollback(clientIdentity, rollback.getBatchId()); // 只回滚单个批次
}
} else {
NettyUtils.error(401,
MessageFormatter.format("destination or clientId is null", rollback.toString())
.getMessage(),
ctx.getChannel(),
null);
}
break;

这里的batchId传入的是0，也就是要回滚所有的批次。我们来看下这个回滚的动作：

@Override
public void rollback(ClientIdentity clientIdentity) throws CanalServerException {
checkStart(clientIdentity.getDestination());
CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
// 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
if (!hasSubscribe) {
return;
}
synchronized (canalInstance) {
// 清除batch信息
canalInstance.getMetaManager().clearAllBatchs(clientIdentity);
// rollback eventStore中的状态信息
canalInstance.getEventStore().rollback();
logger.info("rollback successfully, clientId:{}", new Object[] { clientIdentity.getClientId() });
}
}

这里回滚的，其实是eventStore中的指针，把get的指针设置为之前ack的指针。

二、订阅数据

当client连接server完成后，就需要进行binlog数据的订阅。

public void subscribe() throws CanalClientException {
subscribe(""); // 传递空字符即可
}
public void subscribe(String filter) throws CanalClientException {
int times = 0;
while (times < retryTimes) {
try {
currentConnector.subscribe(filter);
this.filter = filter;
return;
} catch (Throwable t) {
if (retryTimes == -1 && t.getCause() instanceof InterruptedException) {
logger.info("block waiting interrupted by other thread.");
return;
} else {
logger.warn(String.format(
"something goes wrong when subscribing from server: %s",
currentConnector != null ? currentConnector.getAddress() : "null"),
t);
times++;
restart();
logger.info("restart the connector for next round retry.");
}
}
}
throw new CanalClientException("failed to subscribe after " + times + " times retry.");
}

订阅这块的内容不再赘述，在上面的connect过程中有提到。这边还有一个失败重试的机制，当异常不是中断异常的情况下，会重试重启client connector，直到达到了阈值retryTimes。

三、获取数据

在建立连接和进行数据订阅之后，就可以开始进行binlog数据的获取了。主要的方法是getWithOutAck这个方法，这种是需要client自己进行数据ack的，保证了只有数据真正的被消费，而且进行了业务逻辑处理之后，才会ack。当然，如果有了异常，也会进行一定次数的重试和重启。

public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {
waitClientRunning();
try {
...//忽略
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());
return receiveMessages();
} catch (IOException e) {
throw new CanalClientException(e);
}
}

我们可以看到，其实是发送了一个GET命令给server端，然后传递了一个参数batchSize，还有超时时间，而且不是自动提交的。服务端的处理是这样的：

embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize());

也是调用的这个方法：

@Override
public Message getWithoutAck(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit)
throws CanalServerException {
checkStart(clientIdentity.getDestination());
checkSubscribe(clientIdentity);
CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
synchronized (canalInstance) {
// 获取到流式数据中的最后一批获取的位置
PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().getLastestBatch(clientIdentity);
Events<Event> events = null;
if (positionRanges != null) { // 存在流数据
events = getEvents(canalInstance.getEventStore(), positionRanges.getStart(), batchSize, timeout, unit);
} else {// ack后第一次获取
Position start = canalInstance.getMetaManager().getCursor(clientIdentity);
if (start == null) { // 第一次，还没有过ack记录，则获取当前store中的第一条
start = canalInstance.getEventStore().getFirstPosition();
}
events = getEvents(canalInstance.getEventStore(), start, batchSize, timeout, unit);
}
if (CollectionUtils.isEmpty(events.getEvents())) {
logger.debug("getWithoutAck successfully, clientId:{} batchSize:{} but result is null",
clientIdentity.getClientId(), batchSize);
return new Message(-1, new ArrayList<Entry>()); // 返回空包，避免生成batchId，浪费性能
} else {
// 记录到流式信息
Long batchId = canalInstance.getMetaManager().addBatch(clientIdentity, events.getPositionRange());
List<Entry> entrys = Lists.transform(events.getEvents(), new Function<Event, Entry>() {
public Entry apply(Event input) {
return input.getEntry();
}
});
if (logger.isInfoEnabled()) {
logger.info("getWithoutAck successfully, clientId:{} batchSize:{}
real size is {} and result is [batchId:{} , position:{}]",
clientIdentity.getClientId(),
batchSize,
entrys.size(),
batchId,
events.getPositionRange());
}
return new Message(batchId, entrys);
}
}
}

最主要的逻辑在这里：

• 判断canalInstance是否已经启动：checkStart
• 判断订阅列表中是否包含当前的client：checkSubscribe
• 根据client信息从metaManager中获取最后消费的批次：getLastestBatch，这块在运行起来后，是从内存中取的，但是在instance启动时，是从zk中拉取的，是从/otter/canal/destinations/{destination}/{clientId}/mark下面获取的，后续也会定时（1s）刷新到这里面
• 如果能获取到消费的批次，直接从eventStore的队列中获取数据。
• 如果positionRanges为空，那么从metaManager中获取指针。如果指针也没有，说明原来没有ack过数据，需要从store中第一条开始获取。这个过程其实就是找start，也就是上一次ack的位置。
• 调用getEvent，获取数据。根据传入的参数不同，调用不同的方法去获取数据，但是最终都是调用的goGet方法。这个doGet方法不是很复杂，主要是根据参数从store队列中获取数据，然后把指针进行新的设置。
• 如果没有取到binlog数据，那么直接返回，批次号为-1。
• 如果取到了数据，记录一下流式数据后返回。

结果封装在Messages中，最终改为Message，包含批次号和binlog列表。

四、业务处理

拿到message后，需要进行判断batchId，如果batchId=-1或者binlog大小为0，说明没有拿到数据。否则在message基础上进行逻辑处理。

Message的内容，后续我们再进行讨论。

五、提交确认

connector.ack(batchId); // 提交确认

提交批次id，底层发送CLIENTACK命令到server。server调用CanalServerWithEmbedded的ack方法来进行提交。

public void ack(ClientIdentity clientIdentity, long batchId) throws CanalServerException {
checkStart(clientIdentity.getDestination());
checkSubscribe(clientIdentity);
CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
PositionRange<LogPosition> positionRanges = null;
positionRanges = canalInstance.getMetaManager().removeBatch(clientIdentity, batchId); // 更新位置
if (positionRanges == null) { // 说明是重复的ack/rollback
throw new CanalServerException(String.format("ack error , clientId:%s batchId:%d is not exist , please check",
clientIdentity.getClientId(),
batchId));
}
// 更新cursor
if (positionRanges.getAck() != null) {
canalInstance.getMetaManager().updateCursor(clientIdentity, positionRanges.getAck());
if (logger.isInfoEnabled()) {
logger.info("ack successfully, clientId:{} batchId:{} position:{}",
clientIdentity.getClientId(),
batchId,
positionRanges);
}
}
// 可定时清理数据
canalInstance.getEventStore().ack(positionRanges.getEnd());
}

首先更新metaManager中的batch，然后更新ack指针，同时清理store中到ack指针位置的数据。

六、回滚

如果有失败的情况，需要进行回滚。发送CLIENTROLLBACK命令给server端，进行数据回滚。回滚单个批次时的处理逻辑是这样的：

@Override
public void rollback(ClientIdentity clientIdentity, Long batchId) throws CanalServerException {
checkStart(clientIdentity.getDestination());
CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
// 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
if (!hasSubscribe) {
return;
}
synchronized (canalInstance) {
// 清除batch信息
PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().removeBatch(clientIdentity,
batchId);
if (positionRanges == null) { // 说明是重复的ack/rollback
throw new CanalServerException(String.format("rollback error, clientId:%s batchId:%d is not exist , please check",
clientIdentity.getClientId(),
batchId));
}
// lastRollbackPostions.put(clientIdentity,
// positionRanges.getEnd());// 记录一下最后rollback的位置
// TODO 后续rollback到指定的batchId位置
canalInstance.getEventStore().rollback();// rollback
// eventStore中的状态信息
logger.info("rollback successfully, clientId:{} batchId:{} position:{}",
clientIdentity.getClientId(),
batchId,
positionRanges);
}
}

这里的rollback到指定的batchId，其实是假的。他的rollback也是全量回滚到ack的指针位置。

七、断开连接

在发生异常情况时，client会断开与server的连接，也就是disconnect方法。

public void disconnect() throws CanalClientException {
if (rollbackOnDisConnect && channel.isConnected()) {
rollback();
}
connected = false;
if (runningMonitor != null) {
if (runningMonitor.isStart()) {
runningMonitor.stop();
}
} else {
doDisconnnect();
}
}

判断是否在断开连接的时候回滚参数(默认false)和当前socket通道是否连接中，进行回滚。

否则调用runningMonitor.stop方法进行停止。主要的过程是这样的：

• 取消监听/otter/canal/destinations/{destination}/{clientId}/running/节点变化信息
• 删除上面这个节点
• 关闭socket的读通道
• 关闭socket的写通道
• 关闭socket channel