实时数据同步之Maxwell和Canal

🕗 发布于 2024-04-19 16:27 canal maxwell

文章目录

一、概述
二、Maxwell 使用
三、Canal使用

一、概述

1、实时同步工具概述

1.1 Maxwell 概述

官网地址：https://maxwells-daemon.io/

Maxwell 是由美国 Zendesk 开源，用 Java 编写的 MySQL 实时抓取软件。实时读取MySQL 二进制日志 Binlog，并生成 JSON格式的消息，作为生产者发送给 Kafka，Kinesis、RabbitMQ、Redis、Google Cloud Pub/Sub、文件或其它平台的应用程序

注意：1.30.0版本后不在支持JDK8

1.2 Canal概述

Canal 是用 Java 开发的基于数据库增量日志解析，提供增量数据订阅&消费的中间件。目前。Canal 主要支持了 MySQL 的 Binlog 解析，解析完成后才利用 Canal Client 来处理获得的相关数据。（数据库同步需要阿里的 Otter 中间件，基于 Canal）

2、数据同步工作原理

2.1 MySQL 主从复制过程

具体可以参考：ShardingSphere数据库中间件基础学习

Master 主库将改变记录，写到二进制日志(binary log)中
Slave 从库向 mysql master 发送 dump 协议，将 master 主库的 binary log events 拷贝到它的中继日志(relay log)；
Slave 从库读取并重做中继日志中的事件，将改变的数据同步到自己的数据库

2.2 两种工具工作原理

就是把自己伪装成 Slave，假装从 Master 复制数据

3、MySQL 的 binlog详解

3.1 什么是 binlog

MySQL 的二进制日志可以说 MySQL 最重要的日志了，它记录了所有的 DDL 和 DML(除了数据查询语句)语句，以事件形式记录，还包含语句所执行的消耗的时间，MySQL 的二进制日志是事务安全型的。一般来说开启二进制日志大概会有 1%的性能损耗。二进制有两个最重要的使用场景:

MySQL Replication 在 Master 端开启 binlog，Master 把它的二进制日志传递给 slaves 来达到 master-slave 数据一致的目的
自然就是数据恢复了，通过使用 mysqlbinlog 工具来使恢复数据。二进制日志包括两类文件：二进制日志索引文件（文件名后缀为.index）用于记录所有的二进制文件，二进制日志文件（文件名后缀为.00000*）记录数据库所有的 DDL 和 DML(除了数据查询语句)语句事件

3.2 binlog 的开启

找到 MySQL 配置文件的位置
Linux: /etc/my.cnf 如果/etc 目录下没有，可以通过 locate my.cnf 查找位置
Windows: \my.ini
在 mysql 的配置文件下,修改配置在[mysqld] 区块，设置/添加 log-bin=mysql-bin这个表示 binlog 日志的前缀是 mysql-bin，以后生成的日志文件就是 mysql-bin.000001的文件后面的数字按顺序生成，每次 mysql 重启或者到达单个文件大小的阈值时，新生一个文件，按顺序编号

3.3 binlog 的分类设置

mysql binlog 的格式有三种，分别是 STATEMENT,MIXED,ROW。在配置文件中可以选择配binlog_format= statement|mixed|row。Maxwell 想做监控分析，选择 row 格式比较合适

statement

语句级，binlog 会记录每次一执行写操作的语句。相对 row 模式节省空间，但是可能产生不一致性，比如update test set create_date=now();如果用 binlog 日志进行恢复，由于执行时间不同可能产生的数据就不同。优点：节省空间缺点：有可能造成数据不一致。
row

行级， binlog 会记录每次操作后每行记录的变化。优点：保持数据的绝对一致性。因为不管 sql 是什么，引用了什么函数，他只记录执行后的效果。缺点：占用较大空间
mixed

混合级别，statement 的升级版，一定程度上解决了 statement 模式因为一些情况而造成的数据不一致问题。默认还是 statement，在某些情况下，譬如：当函数中包含 UUID() 时；包含 AUTO_INCREMENT 字段的表被更新时；执行 INSERT DELAYED 语句时；用 UDF 时；会按照 ROW 的方式进行处理优点：节省空间，同时兼顾了一定的一致性。缺点：还有些极个别情况依旧会造成不一致，另外 statement 和 mixed 对于需要对binlog 监控的情况都不方便。

4、Maxwell和Canal对比

对比	Canal	Maxwell
语言	java	java
数据格式	格式自由	json
采集数据模式	增量	全量/增量
数据落地	定制	支持 kafka 等多种平台
HA	支持	支持

5、环境安装

需要安装mysql，kafka，zookeeper，具体的可以参考之前的博客文章

这里讲解开启mysql的binLog日志

sudo vim /etc/my.cnf
# 在[mysqld]模块下添加一下内容
[mysqld]
# 主机序列号，每台都要唯一
server_id=1
# 前缀
log-bin=mysql-bin
binlog_format=row
# 针对某个库，不加就是全部，多个库就弄多行
#binlog-do-db=test_maxwell
# binlog-do-db=test_maxwell2
# 这是canal测试
#binlog-do-db=gmall-2021

# 重启 Mysql 服务
sudo systemctl restart mysqld
 
# 登录 mysql 并查看是否修改完成
mysql -uroot -p123456
show variables like '%binlog%';
# 查看下列属性
binlog_format | ROW

# 进入/var/lib/mysql 目录，查看 MySQL 生成的 binlog 文件
cd /var/lib/mysql
sudo ls -l

注：MySQL 生成的 binlog 文件初始大小一定是 154 字节，然后前缀是 log-bin 参数配置的，后缀是默从.000001，然后依次递增。除了 binlog 文件文件以外，MySQL 还会额外生产一个.index 索引文件用来记录当前使用的 binlog 文件

二、Maxwell 使用

1、Maxwell 安装部署

1.1 下载安装

# 因为1.30开始不支持jdk8，所以用这个
wget https://github.com/zendesk/maxwell/releases/download/v1.29.2/maxwell-1.29.2.tar.gz
tar -zxvf maxwell-1.29.2.tar.gz -C /opt/module/

1.2 初始化 Maxwell 元数据库

# 在 MySQL 中建立一个 maxwell 库用于存储 Maxwell 的元数据
mysql -uroot -p123456
CREATE DATABASE maxwell;
# 设置 mysql 用户密码安全级别
set global validate_password_length=4;
set global validate_password_policy=0;
# 分配一个账号可以操作该数据库
GRANT ALL ON maxwell.* TO 'maxwell'@'%' IDENTIFIED BY '123456';
# 分配这个账号可以监控其他数据库的权限
GRANT SELECT,REPLICATION SLAVE,REPLICATION CLIENT ON *.* TO maxwell@'%';
# 刷新 mysql 表权限
flush privileges;

1.3 Maxwell 进程启动

# ============Maxwell 进程启动方式有如下两种========
# 1、使用命令行参数启动 Maxwell 进程
cd /opt/module/maxwell-1.29.2/
# --user  连接 mysql 的用户
# --password  连接 mysql 的用户的密码
# --host mysql 安装的主机名
# --producer  生产者模式(stdout：控制台 kafka：kafka 集群)
bin/maxwell --user='maxwell' --password='123456' --host='hadoop102' --producer=stdout
# 2、修改配置文件，定制化启动 Maxwell 进程 
cp config.properties.example config.properties
vim config.properties
# 修改完成后
bin/maxwell --config ./config.properties

2、Maxwell入门案例

2.1 监控 Mysql 数据并在控制台打印

# 运行 maxwell 来监控 mysql 数据更新
bin/maxwell --user='maxwell' --password='123456' --host='hadoop102' --producer=stdout

创建对应的mysql语句，查看控制台

# 向 mysql 的 test_maxwell 库的 test 表插入一条数据，查看 maxwell 的控制台输出
insert into test  values(1,'aaa');
{
    "database":"test_maxwell", --库名
    "table":"test", --表名
    "type":"insert", --数据更新类型
    "ts":1683960319, --操作时间
    "xid":4335, --操作 id
    "commit":true, --提交成功
    "data":{ --数据
        "id":1,
        "name":"aaa"
    }
}
# 向 mysql 的 test_maxwell 库的 test 表同时插入 3 条数据，控制台出现了 3 条 json日志，说明 maxwell 是以数据行为单位进行日志的采集的
INSERT INTO  test VALUES(2,'bbb'),(3,'ccc'),(4,'ddd');
{"database":"test_maxwell","table":"test","type":"insert","ts":1683960373,"xid":4666,"xoffset":0,"data":{"id":2,"name":"bbb"}}
{"database":"test_maxwell","table":"test","type":"insert","ts":1683960373,"xid":4666,"xoffset":1,"data":{"id":3,"name":"ccc"}}
{"database":"test_maxwell","table":"test","type":"insert","ts":1683960373,"xid":4666,"commit":true,"data":{"id":4,"name":"ddd"}}
update test set name='shawn' where id=1;
{
    "database":"test_maxwell",
    "table":"test",
    "type":"update",
    "ts":1683960396,
    "xid":4737,
    "commit":true,
    "data":{
        "id":1,
        "name":"shawn" --修改后的数据
    },
    "old":{ --修改前的数据
        "name":"aaa"
    }
}
# 删除 test_maxwell 库的 test 表的一条数据，查看 maxwell 的控制台输出
DELETE FROM test WHERE id=1;
{
    "database":"test_maxwell",
    "table":"test",
    "type":"delete",
    "ts":1683960501,
    "xid":5085,
    "commit":true,
    "data":{
        "id":1,
        "name":"shawn"
    }
}

2.2 监控 Mysql 数据输出到 kafka

简单接入

# 启动 zookeeper 和 kafka
jpsall
# windows有个可视化工具，叫做kafka Tool
# https://www.kafkatool.com/download.html

# 启动 Maxwell 监控 binlog
bin/maxwell --user='maxwell' --password='123456' --host='hadoop102' --producer=kafka --kafka.bootstrap.servers=hadoop102:9092   --kafka_topic=maxwell
# 打开 kafka 的控制台的消费者消费 maxwell 主题
# 如果要读取历史数据，需要加上--from-begining
kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic maxwell

然后就是kafka 主题数据的分区控制，在公司生产环境中，我们一般都会用 maxwell 监控多个 mysql 库的数据，然后将这些数据发往 kafka 的一个主题 Topic，并且这个主题也肯定是多分区的，为了提高并发度。那么如何控制这些数据的分区问题，就变得至关重要，实现步骤如下：在公司生产环境中，我们一般都会用 maxwell 监控多个 mysql 库的数据，然后将这些数据发往 kafka 的一个主题 Topic，并且这个主题也肯定是多分区的

# 修改 maxwell 的配置文件，定制化启动 maxwell 进程
vim config.properties

#   tl;dr   config
log_level=info
# #Maxwell数据发送目的地，可选配置有stdout|file|kafka|kinesis|pubsub|sqs|rabbitmq|redis
producer=kafka
kafka.bootstrap.servers=hadoop102:9092
 
#   mysql   login   info
host=hadoop102
user=maxwell
password=123456
jdbc_options=useSSL=false&serverTimezone=Asia/Shanghai
#  ***   kafka   ***
#   list   of   kafka   brokers
#kafka.bootstrap.servers=hosta:9092,hostb:9092
#   kafka   topic   to   write   to
#   this   can   be   static,   e.g.   'maxwell',   or   dynamic,   e.g.namespace_%{database}_%{table}
#   in   the   latter   case   'database'   and   'table'   will   be   replacedwith   the   values   for   the   row   being   processed
# #目标Kafka topic，可静态配置，例如:maxwell，也可动态配置，例如：%{database}_%{table}
kafka_topic=maxwell3

#  ***   partitioning   ***
# 一般都是库名或表分区
#   What   part   of   the   data   do   we   partition   by?
#producer_partition_by=database   #   [database,   table,primary_key,   transaction_id,   column]
producer_partition_by=database

#   控制数据分区模式，可选模式有  库名，表名，主键，列名
#   specify   what   fields   to   partition   by   when   using producer_partition_by=column
#   column   separated   list.
#producer_partition_columns=name
 
#   when   using   producer_partition_by=column,   partition   by   this when
#   the   specified   column(s)   don't   exist.
#producer_partition_by_fallback=database

# 手动创建一个 3 个分区的 topic，名字就叫做 maxwell3
kafka-topics.sh --zookeeper hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka --create --replication-factor 2 --partitions 3 --topic maxwell3
# 利用配置文件启动 Maxwell 进程
bin/maxwell --config ./config.properties
/opt/module/maxwell/bin/maxwell --config /opt/module/maxwell/config.properties --daemon

ps -ef | grep maxwell | grep -v grep | grep maxwell | awk '{print $2}' | xargs kill -9

# 消费者
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic maxwell3

# 向 test_maxwell 库的 test 表再次插入一条数据
insert into test_maxwell.test values (6,'fff');
# 通过 kafka tool 工具查看，此条数据进入了 maxwell3 主题的 1 号分区
# 向 test_maxwell2 库的 aaa 表插入一条数据
# 注意binlog配置文件要监听这个库才行
insert into test_maxwell2.test values (23,'dd');
# 通过 kafka  tool 工具查看，此条数据进入了 maxwell3 主题的 0 号分区,说明库名会对数据进入的分区造成影响

2.3 监控 Mysql 指定表数据输出控制台

运行 maxwell 来监控 mysql 指定表数据更新

# 运行 maxwell 来监控 mysql 指定表数据更新
bin/maxwell --user='maxwell' --password='123456' --host='hadoop102' --filter 'exclude: *.*, include:test_maxwell.test' --producer=stdout
# 注：还可以设置 include:test_maxwell.*，通过此种方式来监控 mysql 某个库的所有表，也就是说过滤整个库

2.4 监控 Mysql 指定表全量数据输出控制台，数据初始化

https://maxwells-daemon.io/bootstrapping/

Maxwell 进程默认只能监控 mysql 的 binlog日志的新增及变化的数据，但是Maxwell 是支持数据初始化的，可以通过修改 Maxwell 的元数据，来对 MySQL 的某张表进行数据初始化，也就是我们常说的全量同步。具体操作步骤如下：需求：将 test_maxwell 库下的 test2 表的四条数据，全量导入到 maxwell 控制台进行打印

# 修改Maxwell的元数据，触发数据初始化机制，在 mysql 的 maxwell 库中 bootstrap表中插入一条数据，写明需要全量数据的库名和表名
insert into maxwell.bootstrap(database_name,table_name) values('test_maxwell','test2');

# 启动 maxwell 进程，此时初始化程序会直接打印 test2 表的所有数据
bin/maxwell --user='maxwell' --password='123456' --host='hadoop102' --producer=stdout

# 当数据全部初始化完成以后，Maxwell 的元数据会变化
# is_complete  字段从 0 变为 1
# start_at  字段从 null 变为具体时间(数据同步开始时间)
# complete_at  字段从 null 变为具体时间(数据同步结束时间)

还有一个方法是使用maxwell-bootstrap脚本，前提是已经启动了maxwell，否则会被阻塞

/opt/module/maxwell/bin/maxwell-bootstrap --database gmall --table user_info --config /opt/module/maxwell/config.properties

# 第一条type为bootstrap-start和最后一条type为bootstrap-complete的数据，是bootstrap开始和结束的标志，不包含数据，中间的type为bootstrap-insert的数据才包含数据。
# 一次bootstrap输出的所有记录的ts都相同，为bootstrap开始的时间

# 采用bootstrap方式同步的输出数据格式如下
{
    "database": "fooDB",
    "table": "barTable",
    "type": "bootstrap-start",
    "ts": 1450557744,
    "data": {}
}
{
    "database": "fooDB",
    "table": "barTable",
    "type": "bootstrap-insert",
    "ts": 1450557744,
    "data": {
        "txt": "hello"
    }
}
{
    "database": "fooDB",
    "table": "barTable",
    "type": "bootstrap-insert",
    "ts": 1450557744,
    "data": {
        "txt": "bootstrap!"
    }
}
{
    "database": "fooDB",
    "table": "barTable",
    "type": "bootstrap-complete",
    "ts": 1450557744,
    "data": {}
}

2.5 群起脚本

一个启动脚本，可以参考

#!/bin/bash

MAXWELL_HOME=/opt/module/maxwell

status_maxwell(){
    result=`ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | wc -l`
    return $result
}


start_maxwell(){
    status_maxwell
    if [[ $? -lt 1 ]]; then
        echo "启动Maxwell"
        $MAXWELL_HOME/bin/maxwell --config $MAXWELL_HOME/config.properties --daemon
    else
        echo "Maxwell正在运行"
    fi
}


stop_maxwell(){
    status_maxwell
    if [[ $? -gt 0 ]]; then
        echo "停止Maxwell"
        ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | awk '{print $2}' | xargs kill -9
    else
        echo "Maxwell未在运行"
    fi
}


case $1 in
    start )
        start_maxwell
    ;;
    stop )
        stop_maxwell
    ;;
    restart )
       stop_maxwell
       start_maxwell
    ;;
esac

三、Canal使用

1、Canal 的下载和安装

1.1 下载安装

# https://github.com/alibaba/canal/releases
wget https://github.com/alibaba/canal/releases/download/canal-1.1.2/canal.deployer-1.1.2.tar.gz
mkdir /opt/module/canal
tar -zxvf canal.deployer-1.1.2.tar.gz -C /opt/module/canal

1.2 mysql创建canal用户

mysql -uroot -p123456
set global validate_password_length=4;
set global validate_password_policy=0;
GRANT SELECT, REPLICATION SLAVE,REPLICATION CLIENT ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal';

# 在gmall-2021库下创建数据表
CREATE TABLE user_info(
`id` VARCHAR(255),
`name` VARCHAR(255),
`sex` VARCHAR(255)
);

1.3 修改 canal.properties 的配置

cd /opt/module/canal/conf
vim canal.properties

#################################################
######### common argument ############# 
#################################################
canal.id = 1
canal.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, RocketMQ
canal.serverMode = tcp
# flush meta cursor/parse position to file

说明：这个文件是 canal 的基本通用配置，canal 端口号默认就是 11111，修改 canal 的输出 model，默认 tcp，改为输出到 kafka

多实例配置如果创建多个实例，通过前面 canal 架构，我们可以知道，一个 canal 服务中可以有多个 instance，conf/下的每一个 example 即是一个实例，每个实例下面都有独立的配置文件。默认只有一个实例 example，如果需要多个实例处理不同的 MySQL 数据的话，直接拷贝出多个 example，并对其重新命名，命名和配置文件中指定的名称一致，然后修改canal.properties 中的 canal.destinations=实例1，实例2，实例3

#################################################
######### destinations ############# 
#################################################
canal.destinations = example

1.4 修改 instance.properties

我们这里只读取一个 MySQL 数据，所以只有一个实例，这个实例的配置文件在conf/example 目录下

cd /opt/module/canal/conf/example
vim instance.properties
# 配置 MySQL 服务器地址
## mysql serverId , v1.0.26+ will autoGen 
canal.instance.mysql.slaveId=20
# enable gtid use true/false
canal.instance.gtidon=false
# position info
canal.instance.master.address=hadoop102:3306

# 配置连接 MySQL 的用户名和密码，默认就是我们前面授权的 canal
# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
canal.instance.defaultDatabaseName =test
# enable druid Decrypt database password
canal.instance.enableDruid=false

2、实时监控测试

2.1 TCP 模式测试

首先创建项目，引入依赖

<dependencies>
    <dependency>
        <groupId>com.alibaba.otter</groupId>
        <artifactId>canal.client</artifactId>
        <version>1.1.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.4.1</version>
    </dependency>
</dependencies>

通用监视类–CanalClient

public class CanalClient {

    public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {

        //TODO 获取连接
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("hadoop102", 11111), "example", "", "");

        while (true) {

            //TODO 连接
            canalConnector.connect();

            //TODO 订阅数据库
            canalConnector.subscribe("gmall-2021.*");

            //TODO 获取数据
            Message message = canalConnector.get(100);

            //TODO 获取Entry集合
            List<CanalEntry.Entry> entries = message.getEntries();

            //TODO 判断集合是否为空,如果为空,则等待一会继续拉取数据
            if (entries.size() <= 0) {
                System.out.println("当次抓取没有数据，休息一会。。。。。。");
                Thread.sleep(1000);
            } else {

                //TODO 遍历entries，单条解析
                for (CanalEntry.Entry entry : entries) {

                    //1.获取表名
                    String tableName = entry.getHeader().getTableName();

                    //2.获取类型
                    CanalEntry.EntryType entryType = entry.getEntryType();

                    //3.获取序列化后的数据
                    ByteString storeValue = entry.getStoreValue();

                    //4.判断当前entryType类型是否为ROWDATA
                    if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {

                        //5.反序列化数据
                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);

                        //6.获取当前事件的操作类型
                        CanalEntry.EventType eventType = rowChange.getEventType();

                        //7.获取数据集
                        List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();

                        //8.遍历rowDataList，并打印数据集
                        for (CanalEntry.RowData rowData : rowDataList) {

                            JSONObject beforeData = new JSONObject();
                            List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                            for (CanalEntry.Column column : beforeColumnsList) {
                                beforeData.put(column.getName(), column.getValue());
                            }

                            JSONObject afterData = new JSONObject();
                            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                            for (CanalEntry.Column column : afterColumnsList) {
                                afterData.put(column.getName(), column.getValue());
                            }

                            //数据打印
                            System.out.println("Table:" + tableName +
                                    ",EventType:" + eventType +
                                    ",Before:" + beforeData +
                                    ",After:" + afterData);
                        }
                    } else {
                        System.out.println("当前操作类型为：" + entryType);
                    }
                }
            }
        }
    }
}

然后在服务端启动canal：bin/startup.sh，修改数据库进行测试

2.2 Kafka 模式测试

修改 canal.properties 中 canal 的输出 model，默认 tcp，改为输出到 kafka

#################################################
######### common argument ############# 
#################################################
canal.id = 1
canal.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, RocketMQ
canal.serverMode = kafka
# flush meta cursor/parse position to fil

修改 Kafka 集群的地址

##################################################
######### MQ #############
##################################################
canal.mq.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092

修改 instance.properties 输出到 Kafka 的主题以及分区数。注意：默认还是输出到指定 Kafka 主题的一个 kafka 分区，因为多个分区并行可能会打乱 binlog 的顺序，如果要提高并行度，首先设置 kafka 的分区数 >1, 然后设置canal.mq.partitionHash 属性

# mq config
canal.mq.topic=canal_test
canal.mq.partitionsNum=1
# hash partition config
#canal.mq.partition=0
#canal.mq.partitionHash=mytest.person:id,mytest.role:id

然后启动测试

cd /opt/module/canal/
bin/startup.sh
# 看到 CanalLauncher 你表示启动成功，同时会创建 canal_test 主题
jps
# 启动 Kafka 消费客户端测试，查看消费情况
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic canal_test
# 向 MySQL 中插入数据后查看消费者控制台
INSERT INTO user_info VALUES('1001','zhangsan','male'),('1002','lisi','female');

原文地址：https://blog.csdn.net/lemon_TT/article/details/137791162

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：js BOM模型常用方法梳理
下一篇：基于单目相机的标靶三维定位——编程实现

spark.default.parallelism 在什么时候起作用，与spark.sql.shuffle.partitions有什么异同点？
异同主要影响 RDD 的初始并行度，而则专门控制 Spark SQL 中的 Shuffle 分区数。前者在非 SQL 的 RDD 操作中起作用，后者则仅对 SQL 或 DataFrame API 中的
阅读更多2024-11-15
QT 控件大小的自适应（重写resizeEvent方法）
【代码】QT 控件大小的自适应（重写resizeEvent方法）
阅读更多2024-11-15
【每日 C/C++ 问题】
当进程需要访问某个页面时，操作系统会将其加载到物理内存中并更新进程的页表，如果物理内存不足，则可能需要将某些不常用的页面置换出来，供其他进程使用。如果程序要访问虚拟地址的时候，由操作系统转换成不同的物
阅读更多2024-11-15
Linux 批量配置互信
linux服务器批量配置互信
阅读更多2024-11-15
数据结构-布隆过滤器和可逆布隆过滤器
在解决缓存穿透问题时，往往会用到一种高效的数据结构-布隆过滤器，其能够快速过滤掉不存在的非法请求，但其也存在一定的误差，即少量不存在的请求也会被放过去。本文对布隆过滤器家族进行介绍，除了常见的普通布隆
阅读更多2024-11-15
【数据结构】AVL树
本文主要介绍了数据结构中的AVL树
阅读更多2024-11-15
Elasticsearch 重建索引&数据迁移
ES 索引不支持直接修改某个字段类型，如果该索引中已经存在数据。如果要修改字段类型，只能通过重建索引+数据迁移的方式来实现该效果。Index Management 页面找到新建的临时索引，点击名称超链
阅读更多2024-11-15
1300. 转变数组后最接近目标值的数组和
给你一个整数数组 arr 和一个目标值 target ，请你返回一个整数 value ，使得将数组中所有大于 value 的值变成 value 后，数组的和最接近 target （最接近表示两者之差
阅读更多2024-11-15
【AI日记】24.11.14 复习和准备 RAG 项目 | JavaScript RAG Web Apps with LlamaIndex
复习和准备 RAG 项目 | JavaScript RAG Web Apps with LlamaIndex
阅读更多2024-11-15
VUE 实现公告无缝循环滚动
VUE 实现公告无缝向左滚动，当公告的宽度大于当前视图宽度的时候再向左滚动，否则不滚动，当鼠标移入时停止滚动，移开时继续滚动，公告获取后台返回的一个数组，将链接转换为HTML。
阅读更多2024-11-15

实时数据同步之Maxwell和Canal

文章目录

一、概述

1、实时同步工具概述

1.1 Maxwell 概述

1.2 Canal概述

2、数据同步工作原理

2.1 MySQL 主从复制过程

2.2 两种工具工作原理

3、MySQL 的 binlog详解

3.1 什么是 binlog

3.2 binlog 的开启

3.3 binlog 的分类设置

4、Maxwell和Canal对比

5、环境安装

二、Maxwell 使用

1、Maxwell 安装部署

1.1 下载安装

1.2 初始化 Maxwell 元数据库

1.3 Maxwell 进程启动

2、Maxwell入门案例

2.1 监控 Mysql 数据并在控制台打印

2.2 监控 Mysql 数据输出到 kafka

2.3 监控 Mysql 指定表数据输出控制台

2.4 监控 Mysql 指定表全量数据输出控制台，数据初始化

2.5 群起脚本

三、Canal使用

1、Canal 的下载和安装

1.1 下载安装

1.2 mysql创建canal用户

1.3 修改 canal.properties 的配置

1.4 修改 instance.properties

2、实时监控测试

2.1 TCP 模式测试

2.2 Kafka 模式测试

相关文章