AI 助理
备案控制台
开发者社区 数据库 文章 正文

canal应用-1个server+2个instance+2个client+2个mysql

2022-05-16 457
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
RDS MySQL Serverless 基础系列，0.5-2RCU 50GB
推荐场景：
学生管理系统数据库设计 搭建个人博客
云数据库 RDS MySQL，集群版 2核4GB 100GB
推荐场景：
搭建个人博客
云数据库 RDS MySQL，高可用版 2核4GB 50GB
简介： canal应用-1个server+2个instance+2个client+2个mysql

一 canal应用架构设计


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



组件说明:


  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. 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 canal server配置与启动


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. 修改配置


新解压的文件夹/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 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 ......


5. 启动canal client


注意运行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 java.net.InetSocketAddress;
import java.util.List;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.Message;
public class CanalClientTest {
    public static void main(String args[]) {
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.175.20", 11111),
                "example1", "", "");//或者example2
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");//订阅所有库下面的所有表
            //connector.subscribe("canal.t_canal");//订阅库canal库下的表t_canal
            connector.rollback();
            int totalEmtryCount = 1200;
            while (emptyCount < totalEmtryCount) {//实际生产中需要设置为true,死循环
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    System.out.println("empty count : " + emptyCount);//此時代表當前數據庫無遍更數據
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    emptyCount = 0;
                    System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                    printEntry(message.getEntries());
                }
                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
            System.out.println("empty too many times, exit");
        } 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(10,'hello',1);

canal client输出日志:

================> binlog[mysql-bin.000001:6764] , name[canal,t_canal] , eventType : INSERT
id : 10    update=true
name : hello    update=true
status : 1    update=true


三. 自问自答-为何设置了数据表的过滤条件，但貌似没有生效？

答：首先看文档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. https://www.cnblogs.com/jayinnn/p/9606466.html
  2. https://github.com/alibaba/canal
文章标签：
云数据库 RDS MySQL 版
Java
消息中间件
分布式计算
SQL
关系型数据库
Linux
canal
MySQL
Hadoop
数据库
关键词：
云数据库 RDS MySQL 版server
云数据库 RDS MySQL 版应用
云数据库 RDS MySQL 版client
云数据库 RDS MySQL 版canal
云数据库 RDS MySQL 版instance
相关实践学习
如何在云端创建MySQL数据库
开始实验后，系统会自动创建一台自建MySQL的 源数据库 ECS 实例和一台 目标数据库 RDS。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心，利用清洁计算为万物互联的新世界提供源源不断的能源动力，目前开服的区域包括中国（华北、华东、华南、香港）、新加坡、美国（美东、美西）、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程，来了解阿里云能够为你的业务带来哪些帮助 &nbsp; &nbsp; 相关的阿里云产品：云服务器ECS 云服务器 ECS（Elastic Compute Service）是一种弹性可伸缩的计算服务，助您降低 IT 成本，提升运维效率，使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
游客tmj4hgt7bfvas
目录
相关文章
VipSoft
|
6天前
|
数据采集 关系型数据库 MySQL
大数据-业务数据采集-FlinkCDC The MySQL server is not configured to use a ROW binlog_format
大数据-业务数据采集-FlinkCDC The MySQL server is not configured to use a ROW binlog_format
VipSoft
16 1
东方睿赢
|
1天前
|
canal 关系型数据库 MySQL
"揭秘阿里数据同步黑科技Canal：从原理到实战，手把手教你玩转MySQL数据秒级同步，让你的数据处理能力瞬间飙升，成为技术界的新晋网红！"
【8月更文挑战第18天】Canal是一款由阿里巴巴开源的高性能数据同步系统，它通过解析MySQL的增量日志（Binlog），提供低延迟、可靠的数据订阅和消费功能。Canal模拟MySQL Slave与Master间的交互协议来接收并解析Binary Log，支持数据的增量同步。配置简单直观，包括Server和Instance两层配置。在实战中，Canal可用于数据库镜像、实时备份等多种场景，通过集成Canal Client可实现数据的消费和处理，如更新缓存或写入消息队列。
东方睿赢
16 0
源码星辰
|
1月前
|
SQL Oracle 关系型数据库
MySQL、SQL Server和Oracle数据库安装部署教程
数据库的安装部署教程因不同的数据库管理系统（DBMS）而异，以下将以MySQL、SQL Server和Oracle为例，分别概述其安装部署的基本步骤。请注意，由于软件版本和操作系统的不同，具体步骤可能会有所变化。
源码星辰
75 3
屿小夏
|
1月前
|
安全 关系型数据库 MySQL
【Python】已解决：pymysql.err.OperationalError：(2003 “Can’t connect to MySQL server on ‘localhost’ ([WinEr
【Python】已解决：pymysql.err.OperationalError：(2003 “Can’t connect to MySQL server on ‘localhost’ ([WinEr
屿小夏
64 1
爱你三千遍斯塔克
|
1月前
|
关系型数据库 MySQL 数据库
2003-Can`t connect to Mysql server on ‘154.8.165.152‘(10038)
2003-Can`t connect to Mysql server on ‘154.8.165.152‘(10038)
爱你三千遍斯塔克
33 0
爱你三千遍斯塔克
|
1月前
|
关系型数据库 MySQL Java
软件开发2003 -Can·t to MySQL server on ‘xxxxxx‘(10038)，宝塔初始化安装mysql，远程链接MySql注意事项，开始时服务器是没有放开端口的，宝塔也都开
软件开发2003 -Can·t to MySQL server on ‘xxxxxx‘(10038)，宝塔初始化安装mysql，远程链接MySql注意事项，开始时服务器是没有放开端口的，宝塔也都开
爱你三千遍斯塔克
26 0
爱你三千遍斯塔克
|
1月前
|
关系型数据库 MySQL 数据库连接
Cant‘t connect to MySQL server on ‘localhost‘ (10038)
Cant‘t connect to MySQL server on ‘localhost‘ (10038)
爱你三千遍斯塔克
25 0
。思索
|
6天前
|
存储 关系型数据库 MySQL
MySQL——数据库备份上传到阿里云OSS存储
MySQL——数据库备份上传到阿里云OSS存储
。思索
20 0
热爱技术的小郑
|
5天前
|
缓存 NoSQL Redis
一天五道Java面试题----第九天（简述MySQL中索引类型对数据库的性能的影响---------＞缓存雪崩、缓存穿透、缓存击穿）
这篇文章是关于Java面试中可能会遇到的五个问题，包括MySQL索引类型及其对数据库性能的影响、Redis的RDB和AOF持久化机制、Redis的过期键删除策略、Redis的单线程模型为何高效，以及缓存雪崩、缓存穿透和缓存击穿的概念及其解决方案。
热爱技术的小郑
28 2
小王老师呀
|
8天前
|
Oracle 关系型数据库 MySQL
Mysql和Oracle数据库死锁查看以及解决
【8月更文挑战第11天】本文介绍了解决MySQL与Oracle数据库死锁的方法。MySQL可通过`SHOW ENGINE INNODB STATUS`查看死锁详情，并自动回滚一个事务解除死锁；也可手动KILL事务。Oracle则通过查询V$LOCK与V$SESSION视图定位死锁，并用`ALTER SYSTEM KILL SESSION`命令终止相关会话。预防措施包括遵循ACID原则、优化索引及拆分大型事务。
小王老师呀
29 3

热门文章

最新文章

  • 1
    RDS for SQL server 空间问题排查汇总
  • 2
    mysql加索引的时候到底会不会锁表.深入解析
  • 3
    k8s-部署实例(mysql+wordpress)
  • 4
    springboot+redis+rabbitmq实现模拟秒杀系统(附带docker安装mysql，rabbitmq，redis教程)
  • 5
    mysql存储过程
  • 6
    收集的mysql热复制的资料与常用语句
  • 7
    修改MYSQL登录密码
  • 8
    【转】源码安装Mysql，补装innodb引擎方法
  • 9
    mysql5.6 主从
  • 10
    MySQL Events学习笔记
  • 1
    自动化备份MySQL数据库并上传到远程FTP服务器
    87
  • 2
    DataWorks操作报错合集之DataWorks在同步mysql时报错Code:[Framework-02]，mysql里面有个json类型字段，是什么原因导致的
    87
  • 3
    【Go语言专栏】使用Go语言连接MySQL数据库
    772
  • 4
    DataWorks产品使用合集之在DataWorks中，如何通过PolarDB for MySQL来查看binlog日志
    74
  • 5
    datagrip连接mysql报错： No appropriate protocol (protocol is disabled or cipher suites are inappropriate
    63
  • 6
    Ubuntu 20.04 + mysql8 安装以及配置大小写不敏感
    260
  • 7
    DataWorks产品使用合集之DataWorks还有就是对于mysql中的表已经存在数据了，第一次全量后面增量同步的步骤如何解决
    53
  • 8
    centos7在线安装jdk1.8+tomcat+mysql8+nginx+docker
    348
  • 9
    DataWorks产品使用合集之在DataWorks中，将数据集成功能将AnalyticDB for MySQL中的数据实时同步到MaxCompute中如何解决
    57
  • 10
    DataWorks产品使用合集之在DataWorks中，将ODPS数据导入到MySQL的步骤如何解决
    62

    • 相关课程

    更多
  • MySQL企业常见架构与调优经验分享
  • 云数据库MySQL版快速上手教程
  • 阿里云云原生数据仓库AnalyticDB MySQL版 使用教程
  • MySQL实战进阶
  • 数据库及SQL/MySQL基础
  • 云数据库MySQL快速入门

    • 相关电子书

    更多
  • One Box： 解读事务与分析一体化数据库 HybridDB for MySQL
  • One Box：解读事务与分析一体化数据库HybridDB for MySQL
  • 如何支撑HTAP场景-HybridDB for MySQL系统架构和技术演进

    • 相关实验场景

    更多
  • MySQL基础-学生管理系统数据库设计
  • 如何快速连接云数据库RDS MySQL
  • MySQL数据库快速部署实践
  • RDS MySQL大促场景下的智能化弹性
  • 如何开通SQL Server的专属集群
  • 如何开通MySQL的专属集群

    • 推荐镜像

    更多
  • mysql
  • mariadb
  • postgresql
    • 下一篇
    通义千问API入门教程