前言

   Canal是阿里巴巴集团提供的开源产品，基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费功能，canal模拟mysql的slave，通过binlog推送实现同步数据，当前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x（来自于官方），以下是官网的一张示意图

一、MySQL数据同步Elasticsearch方案对比

   MySQL 同步数据到 Elasticsearch 通常有 Logstash 和 Canal 两个常用的选择方案。它们各有优缺点，适用于不同的场景。下面分别介绍这两种工具的特点。

Logstash

优点
集成度高：Logstash 是 Elastic Stack (ELK) 的一部分，与 Elasticsearch 集成非常紧密，配置和使用相对简单。
丰富的插件：Logstash 提供了大量的输入、过滤和输出插件，可以轻松实现复杂的数据处理逻辑。
社区支持：Logstash 拥有庞大的社区支持和丰富的文档资源，遇到问题时更容易找到解决方案。
缺点
性能问题：Logstash 在处理大量数据时可能会出现性能瓶颈，尤其是在高并发场景下。
资源消耗：Logstash 对系统资源（CPU、内存）的消耗较大，需要有足够的硬件资源支持。
延迟较高：Logstash 的数据同步延迟相对较高，不适合对实时性要求极高的场景。

Canal

优点
低延迟：Canal 通过监听 MySQL 的 Binlog 日志，实时捕获数据变更，延迟较低，适合对实时性要求较高的场景。
轻量级：Canal 相对轻量级，对系统资源的消耗较小。
灵活配置：Canal 支持多种数据同步方式，可以灵活配置同步策略。
缺点
配置复杂：Canal 的配置相对复杂，需要对 MySQL 的 Binlog 日志有较深的理解。
社区支持：相对于 Logstash，Canal 的社区支持和文档资源较少，遇到问题时可能需要自行解决。
依赖 MySQL：Canal 依赖于 MySQL 的 Binlog 日志，如果 MySQL 配置不当，可能会影响数据同步的稳定性。

（二）、选择建议

1、实时性要求：
如果对数据同步的实时性要求较高，建议使用 Canal。
如果对实时性要求不是特别高，可以考虑使用 Logstash。
2、数据处理复杂度：
如果需要进行复杂的数据处理和转换，建议使用 Logstash。
如果数据处理较为简单，可以考虑使用 Canal。
3、资源限制：
如果系统资源有限，建议使用 Canal。
如果系统资源充足，可以考虑使用 Logstash。
   综上所述，选择哪种工具取决于你的具体需求和场景。如果你需要一个简单、轻量且实时性高的方案，Canal 是更好的选择；如果你需要一个集成度高、功能丰富且社区支持良好的方案，Logstash 是更好的选择。

二、canal 工作原理

   canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal ),canal 解析 binary log 对象(原始为 byte 流)获取数据的变更信息

三、canal安装与配置

    canal安装非常简单，官方github链接地址: canal下载，选择合适的版本下载，我本次使用的是1.17最新版本，版本后续可能还会更新，大家自行选择。其中canal.deployer-1.1.7.tar.gz这个包是Canal 服务器的部署包，包含Canal核心功能，用于连接 MySQL 数据库，订阅和消费 Binlog，并将解析后的数据分发给不同的消费者。我们用这个包就能完成数据实时同步到不同的中间件中，安装这个软件包之后，只需要进行一些基础配置即可启动使用。

（一）安装步骤

    首先在服务器安装目录创建文件夹canal，将canal.deployer-1.1.7.tar.gz上传到服务器安装路径进行解压即可

cd /opt && mkdir canal 
cd /opt/canal
tar -zxvf canal.deployer-1.1.7.tar.gz

（二）配置说明

    进入到conf文件夹下面有个example文件夹，这个文件夹是官方的配置例子，我们要进行binlog监听的mysql配置信息都要在这个文件夹里面的instance.properties配置文件配置，如果我们要根据业务重新命名example文件夹，那么在这个canal.properties配置文件里面的canal.destinations = example配置即可，他默认就是example，本篇文章就是用默认的example演示mysql同步es效果，下面主要对example文件夹下的instance.properties配置文件说明，主要是配置mysql连接信息和同步位点信息。

主要参数说明：
1、canal.instance.master.address=127.0.0.1:3306
这个配置指定了主数据库（通常是MySQL）的地址和端口。127.0.0.1是IP地址，3306是MySQL的端口号。
2、canal.instance.master.journal.name=
该配置用于指定MySQL binlog文件的名称。如果留空，Canal会自动获取最新的binlog文件名。在需要从特定的binlog文件开始同步时才需要配置。
3、canal.instance.master.position=
这个配置项表示binlog文件中的位置偏移量。与journal.name配合使用，用于指定从哪个位置开始读取binlog。如果不指定，默认会从最新的位置开始读取。
4、canal.instance.master.timestamp=
该配置项用于指定一个时间戳，Canal将从这个时间戳之后的日志开始同步。通常用于恢复到某个时间点的状态。如果不指定，则默认从最新的日志开始同步。
5、canal.instance.master.gtid=
这个配置项用于指定全局事务标识符（GTID）。在基于GTID的复制环境中，可以通过设置此参数来指定从哪个GTID开始同步。如果不指定，默认会从最新的GTID开始同步。
6、canal.instance.dbUsername
mysql用户名
7、canal.instance.dbPassword
mysql用户密码

配置文件如下，官方包目录下就是这个文件，大家做个参考

#################################################
## mysql serverId , v1.0.26+ will autoGen
# canal.instance.mysql.slaveId=0

# enable gtid use true/false
canal.instance.gtidon=false

# position info
# mysql连接信息
canal.instance.master.address=127.0.0.1:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=

# rds oss binlog
canal.instance.rds.accesskey=
canal.instance.rds.secretkey=
canal.instance.rds.instanceId=

# table meta tsdb info
canal.instance.tsdb.enable=true
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
#canal.instance.tsdb.dbUsername=canal
#canal.instance.tsdb.dbPassword=canal

#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=

# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==

# table regex
canal.instance.filter.regex=.*\\..*
# table black regex
canal.instance.filter.black.regex=mysql\\.slave_.*
# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch

# mq config
canal.mq.topic=example
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,topic2:mytest2\\..*,.*\\..*
canal.mq.partition=0
# hash partition config
#canal.mq.enableDynamicQueuePartition=false
#canal.mq.partitionsNum=3
#canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6
#canal.mq.partitionHash=test.table:id^name,.*\\..*
#
# multi stream for polardbx
canal.instance.multi.stream.on=false
#################################################

（三）启动Canal服务

    进入到bin文件夹下面，官方已经提供了启停脚本，一看便知，执行startup.sh启动Canal服务。

执行ps命令检查服务情况

ps -ef|grep canal

如下，表示启动成功

四、在springboot中连接canal并同步mysql数据到es

    要想实现mysql同步数据到es，首先mysql需要开启binlog日志功能，关于mysqlbinglog开启大家可以参考我另外一篇文章链接: Mysql/Mariadb开启binlog日志，实际上呢，只需要在springboot项目中引入一个canal客户端的依赖，然后通过客户端连接到canal，进而监听到mysql的binlog，然后对监听到的数据进行后续处理，如同步到es，kafka等其他中间件中。

（一）引入canal客户端依赖

    pom文件中引入以下依赖，加载canal客户端

       <dependency>
            <groupId>com.alibaba.otter</groupId>
            <artifactId>canal.client</artifactId>
            <version>1.1.4</version>
        </dependency>

（二）创建数据表

1、 在开启了binlog的mysql中创建一个人员表people用于演示数据同步，主要有主键，姓名、年龄、出生日期、创刊日期、修改日期字段，后面我会通过对这个表的数据进行增删改来演示同步效果

CREATE TABLE `people`  (
  `id` int NOT NULL AUTO_INCREMENT COMMENT '主键',
  `name` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL COMMENT '姓名',
  `age` int NULL DEFAULT NULL COMMENT '年龄',
  `birth` date NULL DEFAULT NULL COMMENT '出生日期',
  `create_time` datetime NULL DEFAULT NULL COMMENT '创建日期',
  `update_time` datetime NULL DEFAULT NULL COMMENT '修改日期',
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 1 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci COMMENT = '人员表' ROW_FORMAT = Dynamic;

2、同时我也在es上创建了一个people索引用于演示，字段对应了mysql数据库字段

（三）springgboot中使用canal同步数据

   创建一个同步数据的方法，将一个同步数据的方法扔到线程池中执行，同步方法的逻辑步骤如下：
1、连接canal服务端
2、订阅需要坚挺的数据库或者表
3、解析数据
4、处理数据
    详细示例代码如下，关键代码注释我已经写明，其中在解析完毕后同步数据到es中我使用的是spring-boot-starter-data-elasticsearch插件，这个插件也比较好用，配置好后即可对es索引的数据进行curd操作。

package com.boot3.mysqlcanales.controller;
import java.net.InetSocketAddress;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.util.*;
import java.util.concurrent.*;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
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.boot3.mysqlcanales.es.entity.EsPeople;
import com.boot3.mysqlcanales.es.service.PeopleRepositoryService;
import com.google.protobuf.InvalidProtocolBufferException;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class CanalClient {

    @Autowired
    PeopleRepositoryService peopleRepositoryService;
    static  int keepAliveTime = 30;             // 非核心线程的空闲存活时长（分钟）
    static  int queueCapacity = 1;           // 队列最大长度
    public static BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(queueCapacity);
    static ThreadPoolExecutor threadPool;
    static{
        threadPool = new ThreadPoolExecutor(1, 1, keepAliveTime, TimeUnit.MINUTES, queue);
    }
    @PostConstruct
    public  void init() {
        //创建异步任务添加到线程池
        CompletableFuture.runAsync(() -> {
            //执行任务
            mysqlSyncEs();
        }, threadPool);
        mysqlSyncEs();
    }
    public  void mysqlSyncEs() {
        System.out.println("开始启动canal");
        // 创建链接，后两个参数是用户名和密码，在搭建canal server时候如果指定了那么在这里也要填写
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.244.128",
                11111), "example", "", "");
        try {
            connector.connect();
            // 指定订阅的数据库
            //例如：.*\\..*代表订阅所有；test.*代表订阅test库下面的所有表；test\\.(table1|table2)代表订阅test库下面的table1和table2表；(test1|test2)\\..*代表订阅test1库和test2库下面的所有表
            //connector.subscribe(".*\\..*");
            connector.subscribe("test.*");
            //在启动 Canal 客户端时，可以调用 rollback 方法从上次未确认的位置继续消费数据，而不是从头开始消费
//            connector.rollback();
            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(100);
                //返回 -1 时，表示在当前请求中没有获取到新的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                // 判断是否为有变化的数据
                if (batchId == -1 || size == 0) {
                    try {
                        //没有变更，休息一会（根据实际业务情况定）
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    //处理数据
                    try {
                        handleData(message.getEntries());
                        // 提交确认
                        connector.ack(batchId);
                    }catch (Exception e){
                        e.printStackTrace();
                        // 处理失败, 回滚数据
                        connector.rollback(batchId);
                    }

                }
            }
        } finally {
            connector.disconnect();
        }
    }

    private  void handleData(List<Entry> entrys) {
        for (Entry entry : entrys) {
            //ROWDATA：表示具体的行数据变更，包括插入（INSERT）、更新（UPDATE）和删除（DELETE）操作,这也是最常用的类型，通过这个类型来捕捉变化的数据进行同步;
            // TRANSACTIONBEGIN：表示一个事务的开始。通常不需要特别处理，除非需要记录事务的开始时间等信息。
            //TRANSACTIONEND：表示一个事务的结束。通常不需要特别处理，除非需要记录事务的结束时间等信息。
            //HEARTBEAT：表示心跳消息，用于保持连接活跃。通常不需要特别处理，除非需要监控连接状态。
            EntryType entryType = entry.getEntryType();
            //不是ROWDATA，跳过
            if (entryType!= EntryType.ROWDATA) {
                continue;
            }
            RowChange rowChage ;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (InvalidProtocolBufferException e) {
                throw new RuntimeException(e);
            }
            // 获取事件类型
            EventType eventType = rowChage.getEventType();
            //取出数据
            List<RowData> rowDatasList = rowChage.getRowDatasList();
            //遍历数据
            for (RowData rowData :rowDatasList) {
                if (eventType == EventType.DELETE) {
                    System.out.println("类型："+eventType.name());
                    //删除数据
                    delPeople(rowData.getBeforeColumnsList());
                } else if (eventType == EventType.INSERT) {
                    System.out.println("类型："+eventType.name());
                    //新增数据
                    savePeople(rowData.getAfterColumnsList());
                }  else if (eventType == EventType.UPDATE){
                    System.out.println("类型："+eventType.name());
                    //修改数据
                    savePeople(rowData.getAfterColumnsList());
                }
            }
        }
    }

    /**
     * 删除数据
     * @param columnsList
     */
    private void delPeople(List<Column> columnsList) {
        EsPeople esPeople = new EsPeople();
        for (Column column : columnsList) {
            if(column.getName().equals("id")) {
                esPeople.setId(Integer.parseInt(column.getValue()));
            }
        }
        if(esPeople.getId()==null){return;}
        peopleRepositoryService.del(esPeople);
        System.out.println("删除成功");

    }

    /**
     * 保存数据
     * @param columnsList
     */
    private  void savePeople(List<Column> columnsList) {
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        EsPeople esPeople = new EsPeople();
        for (Column column : columnsList) {
            if(column.getName().equals("id")) {
                esPeople.setId(Integer.parseInt(column.getValue()));
            }else if (column.getName().equals("name")) {
                esPeople.setName(column.getValue());
            }else if (column.getName().equals("age")) {
                if(StringUtils.isNotBlank(column.getValue())) {
                    esPeople.setAge(Integer.parseInt(column.getValue()));
                }
            }else if (column.getName().equals("birth")) {
                if(StringUtils.isNotBlank(column.getValue())) {
                    esPeople.setBirth(LocalDate.parse(column.getValue()));
                }
            }else if (column.getName().equals("create_time")) {
                try {
                    if(StringUtils.isNotBlank(column.getValue())) {
                        esPeople.setCreateTime(dateFormat.parse(column.getValue()));
                    }
                } catch (ParseException e) {
                   e.printStackTrace();
                }
            }else if (column.getName().equals("update_time")) {
                try {
                    if(StringUtils.isNotBlank(column.getValue())) {
                        esPeople.setUpdateTime(dateFormat.parse(column.getValue()));
                    }
                } catch (ParseException e) {
                    e.printStackTrace();
                }
            }
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
        peopleRepositoryService.save(esPeople);
        System.out.println("同步成功");
    }
}

1、增加数据，执行sql

insert into people VALUES(null,'胡歌',18,'1983-01-01','2025-01-21 18:59:56','2025-01-21 18:59:56')

控制台日志显示

es中数据，可见一条数据已经同步到es中，其中日期相差8小时是因为es使用的utc时间，我们在程序中查询时候转换一下就可以，

2、修改一条数据，我们把胡歌的年龄改成30岁

update people set age=30 where name='胡歌'

控制台日志显示

es中数据，可见agq字段已经被改成了30

3、删除一条数据，我们把id=1的数据进行删除

delete from people  where id=1

控制台日志显示

es中数据，刷新一看，es中数据已经被删除

五、总结

    通过canal可以实时将mysql数据的变化情况同步到其他中间件，这在解决缓存一致性，和数据处理时效性方面有很大的帮助，大家在实际工作中如果有这方面的需求可以考虑使用canal

为了帮助更多像你一样的读者，我将持续在专栏中分享技术干货和实用技巧。如果你觉得这篇文章对你有帮助，可以考虑关注我的专栏，谢谢。