【源码】canal和otter的高可靠性分析

本文涉及的产品
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
云数据库 RDS MySQL,高可用系列 2核4GB
简介:

一般来说,我们对于数据库最主要的要求就是:数据不丢。不管是主从复制,还是使用类似otter+canal这样的数据库同步方案,我们最基本的需求是,在数据不丢失的前提下,尽可能的保证系统的高可用,也就是在某个节点挂掉,或者数据库发生主从切换等情况下,我们的数据同步系统依然能够发挥它的作用--数据同步。本文讨论的场景是数据库发生主从切换,本文将从源码的角度,来看看otter和canal是如何保证高可用和高可靠的。

一、EventParser

通过阅读文档和源码,我们可以知道,对于一个canal server,基础的框架包括以下几个部分:MetaManager、EventParser、EventSink和EventStore。其中EventParser的作用就是发送dump命令,从mysql数据库获取binlog文件。发送dump命令,可以指定时间戳或者position,从指定的时间或者位置开始dump。我们来看看过程:

首先是CanalServer启动。otter默认使用的是内置版的canal server,所以我们主要看CanalServerWithEmbedded这个类。来看下他的启动过程:

    public void start(final String destination) {
        final CanalInstance canalInstance = canalInstances.get(destination);
        if (!canalInstance.isStart()) {
            try {
                MDC.put("destination", destination);
                canalInstance.start();//启动实例
                logger.info("start CanalInstances[{}] successfully", destination);
            } finally {
                MDC.remove("destination");
            }
        }
    }

我们看下实例启动那一行,跟到AbstractCanalInstance类中

    public void start() {
        super.start();
        if (!metaManager.isStart()) {
            metaManager.start();//源数据管理启动
        }

        if (!alarmHandler.isStart()) {
            alarmHandler.start();//报警处理器启动
        }

        if (!eventStore.isStart()) {
            eventStore.start();//数据存储器启动
        }

        if (!eventSink.isStart()) {
            eventSink.start();//数据过滤器启动
        }

        if (!eventParser.isStart()) {//数据解析器启动
            beforeStartEventParser(eventParser);
            eventParser.start();
            afterStartEventParser(eventParser);
        }
        logger.info("start successful....");
    }

我们主要看下eventParser.start()方法里面的内容。我们主要关注的是EventParser使如何在主从切换的条件下,进行dump节点的确定的。我们跟踪到AbstractEventParser类中的start()方法,重点看下

// 4. 获取最后的位置信息
EntryPosition position = findStartPosition(erosaConnection);

这块有两个实现,但是canal目前使用的是MysqlEventParser,也就是基于Mysql的Binlog文件来进行数据同步。我们看下代码:

protected EntryPosition findStartPosition(ErosaConnection connection) throws IOException {
    EntryPosition startPosition = findStartPositionInternal(connection);
    if (needTransactionPosition.get()) {
        logger.warn("prepare to find last position : {}", startPosition.toString());
        Long preTransactionStartPosition = findTransactionBeginPosition(connection, startPosition);
        if (!preTransactionStartPosition.equals(startPosition.getPosition())) {
            logger.warn("find new start Transaction Position , old : {} , new : {}",
                    startPosition.getPosition(),
                    preTransactionStartPosition);
            startPosition.setPosition(preTransactionStartPosition);
        }
        needTransactionPosition.compareAndSet(true, false);
    }
    return startPosition;
}

对于第一行findStartPositionInternal(connection),我们重点关注的情况是数据库连接地址发生变化,也就是进行了主从切换的情况。

boolean case2 = (standbyInfo == null || standbyInfo.getAddress() == null)
                && logPosition.getPostion().getServerId() != null
                && !logPosition.getPostion().getServerId().equals(findServerId(mysqlConnection));
if (case2) {
    long timestamp = logPosition.getPostion().getTimestamp();
    long newStartTimestamp = timestamp - fallbackIntervalInSeconds * 1000;
    logger.warn("prepare to find start position by last position {}:{}:{}", new Object[]{"", "",
                logPosition.getPostion().getTimestamp()});
    EntryPosition findPosition = findByStartTimeStamp(mysqlConnection, newStartTimestamp);
    // 重新置为一下
    dumpErrorCount = 0;
    return findPosition;
}

我们分析下case2这个条件,其实就是表示的就是配置了主从切换,而且发生了serverId变化的情况,在这种情况下,首先需要获取到事件发生的时间戳,然后将这个事件发生的时间减去60s,也就是向前推一分钟之后,在新的binlog文件中根据新的时间戳来找到当时对应的事件。

这块根据时间戳来寻找事件的过程比较简单,首先根据binglog-index文件找到所有的binlog文件名,然后遍历binlog文件的头,找到binlog文件的写入时间,与新的时间戳进行对比,定位到binlog文件。定位到文件后,直接根据时间戳来进行遍历,找到新的时间戳之前发生的那个事务起始位置。

/**
 * 根据给定的时间戳,在指定的binlog中找到最接近于该时间戳(必须是小于时间戳)的一个事务起始位置。
 * 针对最后一个binlog会给定endPosition,避免无尽的查询
 */
private EntryPosition findAsPerTimestampInSpecificLogFile(MysqlConnection mysqlConnection,
                                                              final Long startTimestamp,
                                                              final EntryPosition endPosition,
                                                              final String searchBinlogFile) {

    final LogPosition logPosition = new LogPosition();
    try {
        mysqlConnection.reconnect();
        // 开始遍历文件
        mysqlConnection.seek(searchBinlogFile, 4L, new SinkFunction<LogEvent>() {

            private LogPosition lastPosition;

            public boolean sink(LogEvent event) {
                EntryPosition entryPosition = null;
                try {
                    CanalEntry.Entry entry = parseAndProfilingIfNecessary(event);
                    if (entry == null) {
                        return true;
                    }

                    String logfilename = entry.getHeader().getLogfileName();
                    Long logfileoffset = entry.getHeader().getLogfileOffset();
                    Long logposTimestamp = entry.getHeader().getExecuteTime();

                    if (CanalEntry.EntryType.TRANSACTIONBEGIN.equals(entry.getEntryType())
                            || CanalEntry.EntryType.TRANSACTIONEND.equals(entry.getEntryType())) {
                        logger.debug("compare exit condition:{},{},{}, startTimestamp={}...", new Object[]{
                                logfilename, logfileoffset, logposTimestamp, startTimestamp});
                        // 事务头和尾寻找第一条记录时间戳,如果最小的一条记录都不满足条件,可直接退出
                        if (logposTimestamp >= startTimestamp) {
                            return false;
                        }
                    }

                    if (StringUtils.equals(endPosition.getJournalName(), logfilename)
                            && endPosition.getPosition() <= (logfileoffset + event.getEventLen())) {
                        return false;
                    }

                    // 记录一下上一个事务结束的位置,即下一个事务的position
                    // position = current +
                    // data.length,代表该事务的下一条offest,避免多余的事务重复
                    if (CanalEntry.EntryType.TRANSACTIONEND.equals(entry.getEntryType())) {
                        entryPosition = new EntryPosition(logfilename,
                                logfileoffset + event.getEventLen(),
                                logposTimestamp);
                        logger.debug("set {} to be pending start position before finding another proper one...",
                                entryPosition);
                        logPosition.setPostion(entryPosition);
                    } else if (CanalEntry.EntryType.TRANSACTIONBEGIN.equals(entry.getEntryType())) {
                        // 当前事务开始位点
                        entryPosition = new EntryPosition(logfilename, logfileoffset, logposTimestamp);
                        logger.debug("set {} to be pending start position before finding another proper one...",
                                entryPosition);
                        logPosition.setPostion(entryPosition);
                    }

                    lastPosition = buildLastPosition(entry);
                } catch (Throwable e) {
                    processSinkError(e, lastPosition, searchBinlogFile, 4L);
                }

                return running;
            }
        });

    } catch (IOException e) {
        logger.error("ERROR ## findAsPerTimestampInSpecificLogFile has an error", e);
    }

    if (logPosition.getPostion() != null) {
        return logPosition.getPostion();
    } else {
        return null;
    }
}

这块的逻辑如下:

  • 发送dump命令,起始位置为4L,也就是跳过了binlog的第一个标志事件。
  • canal收到binlog,开始进行对binlog文件进行解析。
  • 主要我们看的是事务开始和事务提交的事件,判断事务开始或结束的时间,是否小于我们要找的时间戳,如果大于等于,直接遍历下一个事件。
  • 传入了一个endPosition,防止无限扫描。
  • 虽说是从头开始扫描的,但是要想跳出遍历,需要满足一定的条件。在跳出遍历之前,最后一次设置的logPosition才是我们要招的logPosition。
  • 如果是一个事务提交的事件,我们要找的position就是这个事件的position+event.length。如果是事务开始,position就是当前事件的position。其他的事件都忽略。

至此,我们已经找到了我们想要的binlog文件名和对应的事务开始position,我们继续下面的步骤即可。

二、EventStore

这块内容的主要思想如下:

  • 维护一个类似于Disruptor的RingBuffer,同时维护三个序列,put/get/ack。
  • EventSink之后的数据,调用put接口,将数据放入环形队列中。
  • Canal client获取数据,调用get方法。
  • 异步调用ack方法,清除ack之前的数据。
  • 值得注意的是,这块get和ack采用了流式API的模式,get和ack异步进行,可以先get,然后异步调用ack。
  • ack是有序的,不允许跳跃式的提交。

三、Binlog的Row模式

至此,我们基本上知道了canal是如何在发生数据库主从切换时保证高可用和高可靠的,我们可能还有疑惑:为什么要回退60s,来解析binlog,这样不会导致数据重复吗?还有一些自增的update语句(不具备幂等性),不会产生数据错误吗?要想回答这些问题,就需要我们了解Binlog的Row模式了。

Mysql Binlog的Row模式记录的,是数据库中每一行的数据变化,而不仅仅是sql语句。比如我们对数据库中的多行,使用一条sql语句进行了修改。在这种情况下,如果Binlog模式为Statement,只会记录一条sql语句。而Row模式下,会对每一行的数据变化进行记录,以及变化前后每个字段的值。这也就是为什么Row模式的binlog文件如此之大的原因。

对于一些不具备幂等性的sql语句,采用Row语句进行Binlog解析时,也是可以通过重复执行,来保证我们数据的最终一致性的。这也就解释了,为什么要回退60s来进行Binlog位点定位、解析的问题。考虑到Mysql主从的数据复制的延迟性(60s,一般来说的延迟没有这么久),我们可以在主节点挂掉的情况下,回退60s到从节点上继续进行binlog的解析。

当然,也需要考虑一些极端的情况,也就是主从复制确实超过了60s的延迟,在这种情况下,就需要otter登场了。基本思路是:反查数据库同步 (以数据库最新版本同步,解决交替性,比如设置一致性反查数据库延迟阀值为60秒,即当同步过程中发现数据延迟超过了60秒,就会基于PK反查一次数据库,拿到当前最新值进行同步,减少交替性的问题)。

相关实践学习
通过Ingress进行灰度发布
本场景您将运行一个简单的应用,部署一个新的应用用于新的发布,并通过Ingress能力实现灰度发布。
容器应用与集群管理
欢迎来到《容器应用与集群管理》课程,本课程是“云原生容器Clouder认证“系列中的第二阶段。课程将向您介绍与容器集群相关的概念和技术,这些概念和技术可以帮助您了解阿里云容器服务ACK/ACK Serverless的使用。同时,本课程也会向您介绍可以采取的工具、方法和可操作步骤,以帮助您了解如何基于容器服务ACK Serverless构建和管理企业级应用。 学习完本课程后,您将能够: 掌握容器集群、容器编排的基本概念 掌握Kubernetes的基础概念及核心思想 掌握阿里云容器服务ACK/ACK Serverless概念及使用方法 基于容器服务ACK Serverless搭建和管理企业级网站应用
目录
相关文章
|
消息中间件 Kafka Scala
【Kafka从成神到升仙系列 一】Kafka源码环境搭建
【Kafka从成神到升仙系列 一】Kafka源码环境搭建
【Kafka从成神到升仙系列 一】Kafka源码环境搭建
|
canal 关系型数据库 MySQL
Canal服务搭建
Canal服务搭建
1154 1
Canal服务搭建
|
4月前
|
Java API Maven
【zookeeper 第五篇章】Curator 库
Curator 是 Netflix 开源的 ZooKeeper 客户端框架,简化了原生 API 的使用并提供了高级功能。可通过 Maven 添加依赖 `curator-framework` 和 `curator-recipes`。示例代码展示了如何创建 Curator 连接、配置重连策略、进行节点的 CRUD 操作以及事务处理等。例如,使用 `ExponentialBackoffRetry` 实现指数退避重试,通过 `create()` 方法创建持久节点,以及利用 `inTransaction()` 启动事务来保证多个操作的原子性。
102 0
|
1月前
|
canal 监控 关系型数据库
canal的特点是什么?如何使用?
【10月更文挑战第23天】canal的特点是什么?如何使用?
105 3
|
4月前
|
数据采集 分布式计算 Kubernetes
Apache Flink 实践问题之ZooKeeper 网络瞬断时如何解决
Apache Flink 实践问题之ZooKeeper 网络瞬断时如何解决
108 4
|
7月前
|
消息中间件 存储 负载均衡
Kafka 迁移工具 MirrorMaker2 原理起底
MirrorMaker2 (MM2) 是 Kafka 从 2.4.0 开始提供的数据复制工具,用于集群间的数据同步和备份。MM2 有三种部署模式:dedicated mode、standalone mode 和 Kafka Connect mode。AutoMQ即将推出基于MM2的迁移产品,助力用户平滑迁移到AutoMQ平台。
268 2
Kafka 迁移工具 MirrorMaker2 原理起底
|
7月前
|
canal SQL 关系型数据库
Canal入门
Canal入门
205 1
|
canal Java 中间件
总结 canal 使用过程中的几个问题,值得思考一下
在给 canal 分配数据库权限的过程中,由于密码设置的比较简单,会报以下错误 ERROR 1819 (HY000): Your password does not satisfy the current policy requirements
255 0
|
canal SQL 关系型数据库
10.【canal】canal从入门到放弃-mysql+canal+rocketmq实现数据库同步-canal简单使用
【canal】canal从入门到放弃-mysql+canal+rocketmq实现数据库同步-canal简单使用
10.【canal】canal从入门到放弃-mysql+canal+rocketmq实现数据库同步-canal简单使用
|
canal 关系型数据库 MySQL
9.【canal】canal从入门到放弃-mysql+canal+rocketmq实现数据库同步-canal安装
canal从入门到放弃-mysql+canal+rocketmq实现数据库同步-canal安装
9.【canal】canal从入门到放弃-mysql+canal+rocketmq实现数据库同步-canal安装