springboot整合canal数据同步

╰半夏微凉° 2024-04-06 11:29 78阅读 0赞

### Canal介绍 ###

##### 1.1、应用场景 #####

在微服务中，我们采取了feign进行服务间通信，通过访问其他模块的数据来完成当前模块的任务，这样耦合度高，效率相对较低，目前我采取另一种实现方式，通过实时同步数据库表的方式实现，例如我们要统计每天注册与登录人数，我们只需把会员表同步到统计库中，实现本地统计就可以了，这样效率更高，耦合度更低，Canal就是一个很好的数据库同步工具。

##### 1.2、Canal简单介绍 #####

canal是阿里巴巴旗下的一款开源项目，纯Java开发。基于数据库增量日志解析，提供增量数据订阅&消费，目前主要支持了MySQL。

### Canal环境搭建 ###

canal的原理是基于mysql binlog技术，所以这里一定需要开启mysql的binlog写入功能.

##### 2.1.开启mysql服务： #####

service mysql start

或者

systemctl start mysqld.service

##### 2.2、检查binlog功能是否有开启 #####

show variables like 'log_bin';

![在这里插入图片描述][394417eacd78486f8ebd7ec1b1542420.png]

##### 2.3、如果显示状态为OFF表示该功能未开启，开启binlog功能 #####

修改 mysql 的配置文件 my.cnf

vi /etc/my.cnf

追加内容：

#binlog文件名
    log-bin=mysql-bin
    #选择row模式
    binlog_format=ROW
    #mysql实例id,不能和canal的slaveId重复
    server_id=1

##### 2.4、重启 mysql： #####

systemctl restart mysqld.service

##### 2.5、查看是否开启，ON表示开启 #####

![在这里插入图片描述][de41de10c078458599b4b8eece6787d6.png]

##### 2.6、查看是否开启MySQL远程连接 #####

show databases;

use mysql;

select `user`,authentication_string,`Host` from `user`;

![在这里插入图片描述][06c914cf3b5945b8b64b47dad6116eff.png]

##### 2.7、有%号表示开启远程连接，是localhost只能本地连接 #####

设置远程连接，%号即远程连接，IDENTIFIED BY后面跟的自己密码

GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' IDENTIFIED BY 'zzybzb';

刷新，重启

flush privileges;

再次查看

##### 2.8、本地和Linux服务器创建一个相同的数据库和表，用来测试同步数据 #####

![在这里插入图片描述][aaf12fb5335f4979a9dddcea57cfe6d2.png]

### 下载安装canal ###

##### 3.1、下载 #####

下载地址：[https://github.com/alibaba/canal/releases/][https_github.com_alibaba_canal_releases]

##### 3.2、上传到服务器，自己创建个目录来放 #####

![在这里插入图片描述][990a802a3dde4886ad3e324a27e31a52.png]

##### 3.3、解压 #####

![在这里插入图片描述][92610632c9d24df294fe6c4df3f347d2.png]

##### 3.4、修改配置文件 #####

vi conf/example/instance.properties

#需要改成自己的数据库信息
    canal.instance.master.address=192.168.44.132:3306
    
    #需要改成自己的数据库用户名与密码
    canal.instance.dbUsername=root
    canal.instance.dbPassword=zzybzb
    
    #需要改成同步的数据库表规则，例如只是同步一下表
    #canal.instance.filter.regex=.*\\..*
    canal.instance.filter.regex=canal_test.member

##### 3.5、进入bin目录，启动canal #####

./startup.sh

![在这里插入图片描述][12f05341db05401584eb2a4c3b06d8ef.png]

### springboot整合canal ###

##### 4.1、引入依赖 #####

<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>

##### 4.2、application配置文件 #####

# 服务端口
    server.port=10000
    # 服务名
    spring.application.name=canal-client
    
    # 环境设置：dev、test、prod
    spring.profiles.active=dev
    
    # mysql数据库连接
    spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
    spring.datasource.url=jdbc:mysql://localhost:3306/canal_test?serverTimezone=GMT%2B8
    spring.datasource.username=root
    spring.datasource.password=zzybzb

##### 4.3、canalClient类 #####

package com.atguigu.canal.client;
    
    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 com.google.protobuf.InvalidProtocolBufferException;
    import org.apache.commons.dbutils.DbUtils;
    import org.apache.commons.dbutils.QueryRunner;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    import javax.sql.DataSource;
    import java.net.InetSocketAddress;
    import java.sql.Connection;
    import java.sql.SQLException;
    import java.util.Iterator;
    import java.util.List;
    import java.util.Queue;
    import java.util.concurrent.ConcurrentLinkedQueue;
    
    @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.64.131",
                    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);
            }
        }
    }

##### 4.4、启动类 #####

package com.atguigu.canal;
    
    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();
        }
    }

##### 4.5、启动后，Linux服务器的数据库改变，本地的数据也会跟着同步 #####

我启动报错，就没有测试的下文了

### 启动bug ###

Error starting ApplicationContext. To display the conditions report re-run your application with 'debug' enabled.
    2022-09-21 21:44:56.890 ERROR 2300 --- [           main] o.s.boot.SpringApplication               : Application run failed
    
    java.lang.IllegalStateException: Failed to execute CommandLineRunner
    	at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:787) [spring-boot-2.2.1.RELEASE.jar:2.2.1.RELEASE]
    	at org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:768) [spring-boot-2.2.1.RELEASE.jar:2.2.1.RELEASE]
    	at org.springframework.boot.SpringApplication.run(SpringApplication.java:322) [spring-boot-2.2.1.RELEASE.jar:2.2.1.RELEASE]
    	at org.springframework.boot.SpringApplication.run(SpringApplication.java:1226) [spring-boot-2.2.1.RELEASE.jar:2.2.1.RELEASE]
    	at org.springframework.boot.SpringApplication.run(SpringApplication.java:1215) [spring-boot-2.2.1.RELEASE.jar:2.2.1.RELEASE]
    	at com.atguigu.canal.CanalApplication.main(CanalApplication.java:16) [classes/:na]
    Caused by: com.alibaba.otter.canal.protocol.exception.CanalClientException: java.net.ConnectException: Connection refused: connect
    	at com.alibaba.otter.canal.client.impl.SimpleCanalConnector.doConnect(SimpleCanalConnector.java:190) ~[canal.client-1.1.0.jar:na]
    	at com.alibaba.otter.canal.client.impl.SimpleCanalConnector.connect(SimpleCanalConnector.java:114) ~[canal.client-1.1.0.jar:na]
    	at com.atguigu.canal.client.CanalClient.run(CanalClient.java:40) ~[classes/:na]
    	at com.atguigu.canal.CanalApplication.run(CanalApplication.java:22) [classes/:na]
    	at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:784) [spring-boot-2.2.1.RELEASE.jar:2.2.1.RELEASE]
    	... 5 common frames omitted
    Caused by: java.net.ConnectException: Connection refused: connect
    	at sun.nio.ch.Net.connect0(Native Method) ~[na:1.8.0_71]
    	at sun.nio.ch.Net.connect(Net.java:454) ~[na:1.8.0_71]
    	at sun.nio.ch.Net.connect(Net.java:446) ~[na:1.8.0_71]
    	at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:648) ~[na:1.8.0_71]
    	at com.alibaba.otter.canal.client.impl.SimpleCanalConnector.doConnect(SimpleCanalConnector.java:149) ~[canal.client-1.1.0.jar:na]
    	... 9 common frames omitted

##### 5.1、可能原因：防火墙没关 #####

![在这里插入图片描述][c57d7578a72549a9a562aa4172317aaf.png]

##### 5.2、11111端口号没开放 #####

开放端口

firewall-cmd --zone=public --add-port=11111/tcp --permanent

配置立即生效：

firewall-cmd --reload

开放端口存在的问题：FirewallD is not running  
![在这里插入图片描述][b1c796a614024edcbbde504d51a41f4e.png]

开启防火墙：

systemctl start firewalld

##### 5.3、配置文件不对 #####

![在这里插入图片描述][1e5f22730d8c46c2878ec02335fa0f46.png]

bug还没解决，未结束。。。

[394417eacd78486f8ebd7ec1b1542420.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/6037b7c703ed4f3a85ed488c0818e3dc.png
[de41de10c078458599b4b8eece6787d6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/3033f27cf1694ac0930cdada1dafb3d8.png
[06c914cf3b5945b8b64b47dad6116eff.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/68b64348320f47e88ddb4169455c532f.png
[aaf12fb5335f4979a9dddcea57cfe6d2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/b96e3c46f44b4315b6c930253be77682.png
[https_github.com_alibaba_canal_releases]: https://github.com/alibaba/canal/releases/
[990a802a3dde4886ad3e324a27e31a52.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/1c1545fc068c451aa23899142c40cdff.png
[92610632c9d24df294fe6c4df3f347d2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/ecc077a7e474409ba1bf936ad83e7041.png
[12f05341db05401584eb2a4c3b06d8ef.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/808639a650594b559ac363d62e148f1a.png
[c57d7578a72549a9a562aa4172317aaf.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/d1ff802691fa46c2bc64dd44b5f86617.png
[b1c796a614024edcbbde504d51a41f4e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/3eb045ed0a8d463abb66beeac2a52640.png
[1e5f22730d8c46c2878ec02335fa0f46.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/ae162882bbf44c4daddb96884222e463.png