canal环境安装及springboot同步实验

骑猪看日落 2023-01-20 03:58 151阅读 0赞

## 简介 ##

canal是阿里针对mysql数据同步的一个工具，目前只支持mysql数据库。  
官网地址：[https://github.com/alibaba/canal][https_github.com_alibaba_canal]

## 搭建教程 ##

之前我的个人博客讲解过mysql主从的搭建教程，其实canal的原理类似，也是基于mysql的日志分析。如下是官网的原话：

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

搭建的话，需要准备两个mysql数据库，平台无关，两个mysql开启logbin，下载cannal，下载地址：  
[https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz][https_github.com_alibaba_canal_releases_download_canal-1.1.5_canal.deployer-1.1.5.tar.gz]

上传到虚拟机，我在虚拟机上使用docker安装了两个mysql,端口分别为3307和3308  
![在这里插入图片描述][20210511151440754.png]

解压：

tar -zxvf  canal.deployer-1.1.5.tar.gz

编辑conf/example目录下的instance.properties文件，主要修改如下配置：

canal.instance.master.address=192.168.37.129:3307
    canal.instance.dbUsername=canal
    canal.instance.dbPassword=canal

两个mysql创建相同库和相同表结构。  
检查下binlog开启情况。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YW5nYmFva2FuZw_size_16_color_FFFFFF_t_70]

为on才是开启状态  
不是on的需要编辑mysql的配置文件，并重启mysql服务

[mysqld]  
    log-bin=mysql-bin #添加这一行就ok  
    binlog-format=ROW #选择row模式  
    server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复

注意：docker安装的mysql如果没有挂载逻辑卷，需要进入容器进行修改

在3307上创建用户canal

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YW5nYmFva2FuZw_size_16_color_FFFFFF_t_70 1]

一切就绪好后，启动canal服务，执行bin目录下的startup.sh即可

sh startup.sh

启动好canal服务，我们编写程序进行同步（项目经验上经常的操作），即我们开启一个服务  
根据canal.instance.master.address=192.168.37.129:3307可知，3307为master,3308为slave，即我们模拟3307的mysql数据库表数据变动，将自动同步3308的数据。  
application.properties文件

server.port=9988
    spring.application.name=canal-client
    spring.profiles.active=dev
    
    spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
    spring.datasource.url=jdbc:mysql://192.168.37.129:3308/zhanggong_tmp?serverTimezone=GMT%2B8
    spring.datasource.username=root
    spring.datasource.password=123456

pom依赖

<dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
            </dependency>
    
            <dependency>
                <groupId>commons-dbutils</groupId>
                <artifactId>commons-dbutils</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-jdbc</artifactId>
            </dependency>
    
            <dependency>
                <groupId>com.alibaba.otter</groupId>
                <artifactId>canal.client</artifactId>
            </dependency>
        </dependencies>

启动类：

import com.atguigu.canal.client.CanalClient;
    import org.springframework.boot.CommandLineRunner;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    import javax.annotation.Resource;
    
    @SpringBootApplication
    public class CanalApplication implements CommandLineRunner { 
        @Resource
        private CanalClient canalClient;
    
        public static void main(String[] args) { 
            SpringApplication.run(CanalApplication.class, args);
        }
    
        @Override
        public void run(String... strings) throws Exception { 
            //项目启动，执行canal客户端监听
            canalClient.run();
        }
    }

辅助类

import java.net.InetSocketAddress;
    import java.sql.Connection;
    import java.sql.SQLException;
    import java.util.List;
    import java.util.Queue;
    import java.util.concurrent.ConcurrentLinkedQueue;
    
    import javax.annotation.Resource;
    import javax.sql.DataSource;
    
    import org.apache.commons.dbutils.DbUtils;
    import org.apache.commons.dbutils.QueryRunner;
    import org.springframework.stereotype.Component;
    
    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;
    import com.google.protobuf.InvalidProtocolBufferException;
    
    @Component
    public class CanalClient { 
    
        //sql队列
        private Queue<String> SQL_QUEUE = new ConcurrentLinkedQueue<>();
    
        @Resource
        private DataSource dataSource;
    
        /** * canal入库方法 */
        public void run() { 
    
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.37.129",
                    11111), "example", "", "");
            int batchSize = 1000;
            try { 
                connector.connect();
                connector.subscribe(".*\\..*");
                connector.rollback();
                try { 
                    while (true) { 
                        //尝试从master那边拉去数据batchSize条记录，有多少取多少
                        Message message = connector.getWithoutAck(batchSize);
                        long batchId = message.getId();
                        int size = message.getEntries().size();
                        if (batchId == -1 || size == 0) { 
                            Thread.sleep(1000);
                        } else { 
                            dataHandle(message.getEntries());
                        }
                        connector.ack(batchId);
    
                        //当队列里面堆积的sql大于一定数值的时候就模拟执行
                        if (SQL_QUEUE.size() >= 1) { 
                            executeQueueSql();
                        }
                    }
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } catch (InvalidProtocolBufferException e) { 
                    e.printStackTrace();
                }
            } finally { 
                connector.disconnect();
            }
        }
    
        /** * 模拟执行队列里面的sql语句 */
        public void executeQueueSql() { 
            int size = SQL_QUEUE.size();
            for (int i = 0; i < size; i++) { 
                String sql = SQL_QUEUE.poll();
                System.out.println("[sql]----> " + sql);
    
                this.execute(sql.toString());
            }
        }
    
        /** * 数据处理 * * @param entrys */
        private void dataHandle(List<Entry> entrys) throws InvalidProtocolBufferException { 
            for (Entry entry : entrys) { 
                if (EntryType.ROWDATA == entry.getEntryType()) { 
                    RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                    EventType eventType = rowChange.getEventType();
                    if (eventType == EventType.DELETE) { 
                        saveDeleteSql(entry);
                    } else if (eventType == EventType.UPDATE) { 
                        saveUpdateSql(entry);
                    } else if (eventType == EventType.INSERT) { 
                        saveInsertSql(entry);
                    }
                }
            }
        }
    
        /** * 保存更新语句 * * @param entry */
        private void saveUpdateSql(Entry entry) { 
            try { 
                RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                List<RowData> rowDatasList = rowChange.getRowDatasList();
                for (RowData rowData : rowDatasList) { 
                    List<Column> newColumnList = rowData.getAfterColumnsList();
                    StringBuffer sql = new StringBuffer("update " + entry.getHeader().getTableName() + " set ");
                    for (int i = 0; i < newColumnList.size(); i++) { 
                        sql.append(" " + newColumnList.get(i).getName()
                                + " = '" + newColumnList.get(i).getValue() + "'");
                        if (i != newColumnList.size() - 1) { 
                            sql.append(",");
                        }
                    }
                    sql.append(" where ");
                    List<Column> oldColumnList = rowData.getBeforeColumnsList();
                    for (Column column : oldColumnList) { 
                        if (column.getIsKey()) { 
                            //暂时只支持单一主键
                            sql.append(column.getName() + "=" + column.getValue());
                            break;
                        }
                    }
                    SQL_QUEUE.add(sql.toString());
                }
            } catch (InvalidProtocolBufferException e) { 
                e.printStackTrace();
            }
        }
    
        /** * 保存删除语句 * * @param entry */
        private void saveDeleteSql(Entry entry) { 
            try { 
                RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                List<RowData> rowDatasList = rowChange.getRowDatasList();
                for (RowData rowData : rowDatasList) { 
                    List<Column> columnList = rowData.getBeforeColumnsList();
                    StringBuffer sql = new StringBuffer("delete from " + entry.getHeader().getTableName() + " where ");
                    for (Column column : columnList) { 
                        if (column.getIsKey()) { 
                            //暂时只支持单一主键
                            sql.append(column.getName() + "=" + column.getValue());
                            break;
                        }
                    }
                    SQL_QUEUE.add(sql.toString());
                }
            } catch (InvalidProtocolBufferException e) { 
                e.printStackTrace();
            }
        }
    
        /** * 保存插入语句 * * @param entry */
        private void saveInsertSql(Entry entry) { 
            try { 
                RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                List<RowData> rowDatasList = rowChange.getRowDatasList();
                for (RowData rowData : rowDatasList) { 
                    List<Column> columnList = rowData.getAfterColumnsList();
                    StringBuffer sql = new StringBuffer("insert into " + entry.getHeader().getTableName() + " (");
                    for (int i = 0; i < columnList.size(); i++) { 
                        sql.append(columnList.get(i).getName());
                        if (i != columnList.size() - 1) { 
                            sql.append(",");
                        }
                    }
                    sql.append(") VALUES (");
                    for (int i = 0; i < columnList.size(); i++) { 
                        sql.append("'" + columnList.get(i).getValue() + "'");
                        if (i != columnList.size() - 1) { 
                            sql.append(",");
                        }
                    }
                    sql.append(")");
                    SQL_QUEUE.add(sql.toString());
                }
            } catch (InvalidProtocolBufferException e) { 
                e.printStackTrace();
            }
        }
    
        /** * 入库 * @param sql */
        public void execute(String sql) { 
            Connection con = null;
            try { 
                if(null == sql) return;
                con = dataSource.getConnection();
                QueryRunner qr = new QueryRunner();
                int row = qr.execute(con, sql);
                System.out.println("update: "+ row);
            } catch (SQLException e) { 
                e.printStackTrace();
            } finally { 
                DbUtils.closeQuietly(con);
            }
        }
    }

## 验证效果 ##

我们在zhanggong\_tmp数据库的test\_canal表记录增加数据，可以看到程序控制台输出  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YW5nYmFva2FuZw_size_16_color_FFFFFF_t_70 2]

如上，说明使用canal数据同步成功。

## 遇到的问题 ##

插入中文乱码，需要针对mysql设置utf8的编码。  
因为配置的为UTF-8  
![在这里插入图片描述][20210512150340291.png]  
my.cnf

[mysqld]
    character-set-server=utf8
    collation-server=utf8_general_ci
     
    [client]
    default-character-set=utf8
     
    [mysql]
    default-character-set=utf8

前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住分享一下给大家。[点击跳转到教程][Link 1]  
博客若对你有用，欢迎扫码打赏，你的打赏是我写作的动力。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YW5nYmFva2FuZw_size_16_color_FFFFFF_t_70_pic_center]  
欢迎加入康哥IT粉丝交流QQ群833249482  
![在这里插入图片描述][20210511091712437.png_pic_center]

[https_github.com_alibaba_canal]: https://github.com/alibaba/canal
[https_github.com_alibaba_canal_releases_download_canal-1.1.5_canal.deployer-1.1.5.tar.gz]: https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz
[20210511151440754.png]: /images/20221021/95f57a2d41d649aaaf4f4dcef785fd31.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YW5nYmFva2FuZw_size_16_color_FFFFFF_t_70]: /images/20221021/19b8f816ccdc42cb8dcdfbf4992b0161.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YW5nYmFva2FuZw_size_16_color_FFFFFF_t_70 1]: /images/20221021/2433f7bebf5a47498f240e3a6da7db3a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YW5nYmFva2FuZw_size_16_color_FFFFFF_t_70 2]: /images/20221021/c21d097a6d254b95b1ddd35a774dce17.png
[20210512150340291.png]: /images/20221021/31129f4902264e859f27287f3d1e3956.png
[Link 1]: https://www.captainai.net/flyop007/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YW5nYmFva2FuZw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221021/d0d06b51d60e4ea092e4d5cd788903f3.png
[20210511091712437.png_pic_center]: /images/20221021/7d800d6d52514f608b59751eae719869.png