canal之高可用架构设计与应用

本文涉及的产品
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
容器服务 Serverless 版 ACK Serverless,952元额度 多规格
简介: canal之高可用架构设计与应用

一 高可用架构设计



网络异常,图片无法展示
|


架构设计图


配置说明:

zookeeper x 3 + canal x 2 + mysql x 2


组件说明:

  1. linux内核版本(CentOS Linux 7):(命令:uname -a)
    Linux slave1 3.10.0-693.el7.x86_64 #1 SMP Tue Aug 22 21:09:27 UTC 2017 x86_64 x86_64 x86_64 GNU/Linux
  2. mysql版本:(SQL命令:select version(); 或 status)
    Server version: 5.6.43-log MySQL Community Server (GPL)
  3. canal版本:canal-1.1.3
  4. zookeeper版本:zookeeper-3.4.5-cdh5.7.0
  5. JDK版本: 1.8


canal工作原理:

  1. 模拟mysql slave的交互协议,伪装自己为mysql slave,向mysql master发送dump协议;
  2. mysql master收到dump请求,开始推送binary log给slave(也就是canal);
  3. 解析binary log对象(原始为byte流)

了解更多详细更新可以查看文章:【了解canal,看这个就够了】


二 配置与部署流程



2.1 安装mysql数据库


1. 下载安装

在192.168.175.21和192.168.175.22两台服务器上分别安装mysql,具体安装流程可参考文章:Linux-安装MySQL.


2. 创建canal账户

在创建root账号并设置远程访问之后,接着创建canal账号并设置远程访问和权限:

mysql> CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
mysql> GRANT ALL ON canal.* TO 'canal'@'%';
mysql> GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'canal'@'%';
mysql>FLUSH PRIVILEGES;

3. 验证登录

#远程登录
mysql -h 192.168.175.22 -P 3306 -u canal -pcanal
#本地登录
mysql -ucanal -pcanal

4. 修改my.cnf配置(这一步非常关键!!!)


分别在175.21和175.22两台服务器修改my.conf配置,查找my.cnf配置位置命令:whereis my.


示例,在192.168.175.21的my.cnf配置新增如下内容:

log_bin=mysql-bin  #指定bin-log的名称,尽量可以标识业务含义
binlog_format=row  #选择row模式,必须!!!
server_id=1  #mysql服务器id


2.2 搭建zookeeper集群


搭建zookeeper集群地址为192.168.175.20:2181,192.168.175.21:2181,192.168.175.22:2181,具体搭建流程,可查看文章【Zookeepr3.4.5集群搭建】


2.3 搭建canal server集群


前提: mysql已打开binlog功能,且配置binlog模式为row.


1. 下载最新canal安装包

下载地址: https://github.com/alibaba/canal/releases/download/canal-1.1.3/canal.deployer-1.1.3.tar.gz


2.上传并解压

进入192.168.175.20服务器,使用rz命令上传,使用如下命令进行解压至/usr/local/hadoop/app/canal:

tar xzvf canal.deployer-1.1.3.tar.gz -C canal

3. 修改配置instance.properties

新解压的文件夹/usr/local/hadoop/app/canal/conf/有一个example文件夹,一个example就代表一个instance实例.而一个instance实例就是一个消息队列,所以这里可以将文件名改为example1,同时再复制出来一个叫example2.(命名可以使用监听的数据库名)


修改/usr/local/hadoop/app/canal/conf/example1/instance.properties配置文件:

canal.instance.master.address=192.168.175.21:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
canal.mq.topic=example1


修改/usr/local/hadoop/app/canal/conf/example2/instance.properties配置文件:

canal.instance.master.address=192.168.175.22:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
canal.mq.topic=example2


配置文件参数说明,可查看:https://github.com/alibaba/canal/wiki/AdminGuide


4. 修改配置canal.properties

配置/usr/local/hadoop/app/canal/conf/canal.properties是一个对应canal server的全局配置(instance.properties是对应canal instance的配置)。

canal.id = 2  #保证每个canal server的id不同
canal.port = 11111
canal.zkServers =192.168.175.20:2181,192.168.175.21:2181,192.168.175.22:2181
canal.instance.global.spring.xml = classpath:spring/default-instance.xml
#其他配置默认即可.


注意: 两台机器上的instance目录的名字需要保证完全一致,HA模式是依赖于instance name进行管理,同时必须都选择default-instance.xml配置。

配置完成,将文件从192.168.175.20远程复制一份到192.168.175.22上:

#需要确保已开通免密
scp -rp /usr/local/hadoop/app/canal slave2:/usr/local/hadoop/app/

5. 启动canal server

进入文件夹/usr/local/hadoop/app/canal/bin执行如下启动命令:

./startup.sh


查看日志/usr/local/hadoop/app/canal/logs/canal/canal.log,出现如下内容,即表示启动成功:

2019-06-07 21:15:03.372 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## load canal configurations
2019-06-07 21:15:03.427 [main] INFO  c.a.o.c.d.monitor.remote.RemoteConfigLoaderFactory - ## load local canal configurations
2019-06-07 21:15:03.529 [main] INFO  com.alibaba.otter.canal.deployer.CanalStater - ## start the canal server.
2019-06-07 21:15:06.251 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[192.168.175.22:11111]
2019-06-07 21:15:22.245 [main] INFO  com.alibaba.otter.canal.deployer.CanalStater - ## the canal server is running now ......


在zk集群中查看canal节点注册情况:

[zk: localhost:2181(CONNECTED) 27] ls2 /otter/canal/destinations
[example2, example1]
[zk: localhost:2181(CONNECTED) 26] ls2 /otter/canal/cluster
[192.168.175.22:11111, 192.168.175.20:11111]


可以看到canal server节点已经在zk集群上注册成功.

当停掉一个canal server时,可以看到zk上对应的临时节点也会删除.


2.4 使用canal client通过zookeeper连接canal server集群


注意运行canal客户端代码时,一定要先启动canal server!!!

(1) 添加pom依赖

<!--canal-->
    <dependency>
      <groupId>com.alibaba.otter</groupId>
      <artifactId>canal.client</artifactId>
      <version>1.1.3</version>
    </dependency>


(2) canal client代码:

package com.xgh.canal;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import java.net.InetSocketAddress;
import java.util.List;
public class TestCanalByZk {
    public static void main(String args[]) {
        String zkHost="192.168.175.20:2181,192.168.175.21:2181,192.168.175.22:2181";
        // 创建链接
        CanalConnector connector = CanalConnectors.newClusterConnector(zkHost,"example1","","");
        /*CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.175.22", 11111),
                "example", "", "");*/
        int batchSize = 1000;
        int emptyCount = 0;
        long batchId = 0;
        //外层死循环:在canal节点宕机后,抛出异常,等待zk对canal处理切换,切换完后,继续创建连接处理数据
        while(true) {
            try {
                connector.connect();
                connector.subscribe(".*\\..*");//订阅所有库下面的所有表
                //connector.subscribe("canal.t_canal");//订阅库canal库下的表t_canal
                connector.rollback();
                //内层死循环:按频率实时监听数据变化,一旦收到变化数据,立即做消费处理,并ack,考虑消费速度,可以做异步处理并ack.
                while (true) {
                    Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                    batchId = message.getId();
                    int size = message.getEntries().size();
                    //// 偏移量不等于-1 或者 获取的数据条数不为0 时,认为拿到消息,并处理
                    if (batchId == -1 || size == 0) {
                        emptyCount++;
                        System.out.println("empty count : " + emptyCount);//此時代表當前數據庫無遍更數據
                        Thread.sleep(200); //200ms拉一次变动数据
                    } else {
                        emptyCount = 0;
                        System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                        printEntry(message.getEntries());
                    }
                    connector.ack(batchId); // 提交确认
                }
            }catch(Exception e){
                e.printStackTrace();
                connector.rollback(batchId); // 处理失败, 回滚数据
            } finally {
                connector.disconnect();
            }
        }
    }
    private static void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }
            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }
            System.out.println("rowChare ======>"+rowChage.toString());
            EventType eventType = rowChage.getEventType(); //事件類型,比如insert,update,delete
            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(),//mysql的my.cnf配置中的log-bin名稱
                    entry.getHeader().getLogfileOffset(), //偏移量
                    entry.getHeader().getSchemaName(),//庫名
                    entry.getHeader().getTableName(), //表名
                    eventType));//事件名
            for (RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == EventType.DELETE) {
                    printColumn(rowData.getBeforeColumnsList());
                } else if (eventType == EventType.INSERT) {
                    printColumn(rowData.getAfterColumnsList());
                } else {
                    System.out.println("-------> before");
                    printColumn(rowData.getBeforeColumnsList());
                    System.out.println("-------> after");
                    printColumn(rowData.getAfterColumnsList());
                }
            }
        }
    }
    private static void printColumn(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }
}


canal client运行实例:

empty count : 1
empty count : 2
empty count : 3
empty count : 4

6. 触发数据库变更

创建库:create database canal;

创建表:create table t_canal (id int,name varchar(20),status int);

插入数据:insert into t_canal values(11,'xxiao',1);

canal client输出日志:

================> binlog[mysql-bin.000001:6973] , name[canal,t_canal] , eventType : INSERT
id : 11    update=true
name : xxiao    update=true
status : 1    update=true

7. 其他

将canal client代码CanalConnectors.newClusterConnector(zkHost,"example1","","");中的队列名example1换成example2再执行,就可以监听example2对应数据变化了.


8. 问题:为何设置了数据表的过滤条件,但貌似没有生效?


:首先看文档AdminGuide,了解canal.instance.filter.regex的书写格式。mysql 数据解析关注的表,Perl正则表达式.多个正则之间以逗号(,)分隔,转义符需要双斜杠(\) 。常见例子:

  1. 所有表:.*   or  .\..
  2. canal schema下所有表: canal\..*
  3. canal下的以canal打头的表:canal\.canal.*
  4. canal schema下的一张表:canal.test1
  5. 多个规则组合使用:canal\..*,mysql.test1,mysql.test2 (逗号分隔)


检查binlog格式,过滤条件只针对row模式的数据有效(ps. mixed/statement因为不解析sql,所以无法准确提取tableName进行过滤)。

检查下CanalConnector是否调用subscribe(filter)方法;有的话,filter需要和instance.properties的canal.instance.filter.regex一致,否则subscribe的filter会覆盖instance的配置,如果subscribe的filter是...,那么相当于你消费了所有的更新数据

特别注意


三 运行测试及总结



1. 启动两个监听example1的canal client,启动两个监听example2的canal client:

在example1或example2对应的数据发生变化时,两个canal client只有一个消费消息。

当两个监听同一个队列的canal client有一个宕掉时,再有数据变化时,剩下的一个canal client就会开始消费数据。

这就验证了canal client的HA机制:为了保证有序性,一份instance同一时间只能由一个canal client进行get/ack/rollback操作,否则客户端接收无法保证有序.


2. 启动两个canal server并在zk上注册

当停掉其中一个canal server时,当产生数据变化时,整个canal server集群仍可以正常对外提供服务。

这就验证了canal server的HA机制:为了减少对mysql dump的请求,不同server上的instance要求同一时间只能有一个处于running,其他的处于standby状态.


3. 在canal server切换过程中,canal client存在重复消费数据的问题

这点需要在消费端自行进行处理。


参考文章:
  1. https://www.2cto.com/database/201609/547661.html
  2. https://www.cnblogs.com/yulu080808/p/8819260.html
  3. https://github.com/alibaba/canal
  4. https://blog.csdn.net/my201110lc/article/details/78836270
相关实践学习
通过Ingress进行灰度发布
本场景您将运行一个简单的应用,部署一个新的应用用于新的发布,并通过Ingress能力实现灰度发布。
容器应用与集群管理
欢迎来到《容器应用与集群管理》课程,本课程是“云原生容器Clouder认证“系列中的第二阶段。课程将向您介绍与容器集群相关的概念和技术,这些概念和技术可以帮助您了解阿里云容器服务ACK/ACK Serverless的使用。同时,本课程也会向您介绍可以采取的工具、方法和可操作步骤,以帮助您了解如何基于容器服务ACK Serverless构建和管理企业级应用。 学习完本课程后,您将能够: 掌握容器集群、容器编排的基本概念 掌握Kubernetes的基础概念及核心思想 掌握阿里云容器服务ACK/ACK Serverless概念及使用方法 基于容器服务ACK Serverless搭建和管理企业级网站应用
相关文章
|
12天前
|
存储 SQL 关系型数据库
Mysql高可用架构方案
本文阐述了Mysql高可用架构方案,介绍了 主从模式,MHA模式,MMM模式,MGR模式 方案的实现方式,没有哪个方案是完美的,开发人员在选择何种方案应用到项目中也没有标准答案,合适的才是最好的。
69 3
Mysql高可用架构方案
|
15天前
|
运维 持续交付 开发工具
深入浅出:GitOps在微服务架构中的应用
【10月更文挑战第26天】本文深入探讨了GitOps在微服务架构中的应用,介绍了其核心理念、自动化部署流程和增强的可观测性。通过实例展示了GitOps如何简化服务部署、配置管理和故障恢复,并推荐了一些实用工具和开发技巧。
|
6天前
|
监控 Go API
Go语言在微服务架构中的应用实践
在微服务架构的浪潮中,Go语言以其简洁、高效和并发处理能力脱颖而出,成为构建微服务的理想选择。本文将探讨Go语言在微服务架构中的应用实践,包括Go语言的特性如何适应微服务架构的需求,以及在实际开发中如何利用Go语言的特性来提高服务的性能和可维护性。我们将通过一个具体的案例分析,展示Go语言在微服务开发中的优势,并讨论在实际应用中可能遇到的挑战和解决方案。
|
6天前
|
网络协议 数据挖掘 5G
适用于金融和交易应用的低延迟网络:技术、架构与应用
适用于金融和交易应用的低延迟网络:技术、架构与应用
31 5
|
7天前
|
Go 数据处理 API
Go语言在微服务架构中的应用与优势
本文摘要采用问答形式,以期提供更直接的信息获取方式。 Q1: 为什么选择Go语言进行微服务开发? A1: Go语言的并发模型、简洁的语法和高效的编译速度使其成为微服务架构的理想选择。 Q2: Go语言在微服务架构中有哪些优势? A2: 主要优势包括高性能、高并发处理能力、简洁的代码和强大的标准库。 Q3: 文章将如何展示Go语言在微服务中的应用? A3: 通过对比其他语言和展示Go语言在实际项目中的应用案例,来说明其在微服务架构中的优势。
|
5天前
|
监控 持续交付 Docker
Docker 容器化部署在微服务架构中的应用有哪些?
Docker 容器化部署在微服务架构中的应用有哪些?
|
5天前
|
监控 持续交付 Docker
Docker容器化部署在微服务架构中的应用
Docker容器化部署在微服务架构中的应用
|
13天前
|
机器学习/深度学习 人工智能 自然语言处理
医疗行业的语音识别技术解析:AI多模态能力平台的应用与架构
AI多模态能力平台通过语音识别技术,实现实时转录医患对话,自动生成结构化数据,提高医疗效率。平台具备强大的环境降噪、语音分离及自然语言处理能力,支持与医院系统无缝集成,广泛应用于门诊记录、多学科会诊和急诊场景,显著提升工作效率和数据准确性。
|
13天前
|
JavaScript 持续交付 Docker
解锁新技能:Docker容器化部署在微服务架构中的应用
【10月更文挑战第29天】在数字化转型中,微服务架构因灵活性和可扩展性成为企业首选。Docker容器化技术为微服务的部署和管理带来革命性变化。本文探讨Docker在微服务架构中的应用,包括隔离性、可移植性、扩展性、版本控制等方面,并提供代码示例。
51 1
|
15天前
|
运维 Serverless 数据处理
Serverless架构通过提供更快的研发交付速度、降低成本、简化运维、优化资源利用、提供自动扩展能力、支持实时数据处理和快速原型开发等优势,为图像处理等计算密集型应用提供了一个高效、灵活且成本效益高的解决方案。
Serverless架构通过提供更快的研发交付速度、降低成本、简化运维、优化资源利用、提供自动扩展能力、支持实时数据处理和快速原型开发等优势,为图像处理等计算密集型应用提供了一个高效、灵活且成本效益高的解决方案。
51 1