MySql自动同步主库数据（Canal）

客官°小女子只卖身不卖艺 2023-01-02 11:25 155阅读 0赞

# MySql自动同步主库数据（Canal） #

[上篇文章][Link 1]介绍了MongoDB数据库模拟从库实现主从复制效果，不会影响线上数据，本文介绍MySql模拟从库实现主从复制。

*  **Find Action Ctrl+Shift+A，在此对话框中输入想操作的英文立即出现想的操作**
 *  **System.out.println输入sout即可**
 *  **for循环N输入N.for即可**
 *  **return n;输入n.return即可**
 *  **离线写博客**
 *  **导入导出Markdown文件**
 *  **丰富的快捷键**

--------------------

## MySql主从复制原理 ##

> MySql主从复制指数据从一个MySql复制到一个或多个MySql服务器中，基于binlog采用异步复制直接IO，这样数据访问不用一直访问主库来完成，主库负责修改，从库负责读取，实现读写分离。

*  master会将变更数据存入binlog文件中。
 *  slave连接主库时主库会开启一个dump线程发送binlog内容。
 *  slave会启动一个I\\O Thread请求主库发送binlog记录并保存到中继日志中。
 *  从库启动Sql Thread线程读取中继日志，本地重放，使得数据与主库保持一致，最后I/O Thread和SQL Thread将进入睡眠状态，等待下一次被唤醒。

总结：

*  从库生成两个线程I/O Thread、Sql Thread。
 *  I/O Thread请求主库binlog并写到中继日志（relay-log）中。
 *  主库生成一个dump线程，给从库传binlog。
 *  Sql线程读取中继日志解析并执行成sql

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

## Canal原理 ##

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

下载源码修改：conf/example/instance.properties文件

//设置主库IP和端口
    canal.instance.master.address=IP:端口
    //设置同步开始时间（毫秒），可忽略
    canal.instance.master.timestamp=

启动canal：/bin/startup.sh

本文用的阿里云服务，不需要有多过设置，其它小伙伴参照官网配置一下即可。

## 代码 ##

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

SimpleCanalClientExample.class

package canal;
    
    import com.alibaba.otter.canal.client.CanalConnector;
    import com.alibaba.otter.canal.client.CanalConnectors;
    import com.alibaba.otter.canal.common.utils.AddressUtils;
    import com.alibaba.otter.canal.protocol.CanalEntry;
    import com.alibaba.otter.canal.protocol.CanalEntry.Column;
    import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
    import com.alibaba.otter.canal.protocol.Message;
    import org.springframework.util.StringUtils;
    
    import java.net.InetSocketAddress;
    import java.sql.Connection;
    import java.sql.DriverManager;
    import java.sql.SQLException;
    import java.sql.Statement;
    import java.util.ArrayList;
    import java.util.List;
    
    /** * @Author: LailaiMonkey * @Description： * @Date：Created in 2021-01-04 11:53 * @Modified By： */
    public class SimpleCanalClientExample { 
    
        static Statement statement = null;
    
        public static void main(String args[]) { 
    
            //获得存储sqlStatement
            getStatement();
    
            // 创建Canal链接
            CanalConnector connector = getCanalConnector();
            int batchSize = 10000;
            while (true) { 
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(batchSize);
    
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) { 
                    try { 
                        Thread.sleep(1000);
                    } catch (InterruptedException ignored) { 
                    }
                } else { 
                    findEntry(message.getEntries());
                }
    
                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
        }
    
        private static CanalConnector getCanalConnector() { 
            // 创建链接
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalIp,
                    11111), "example", "", "");
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            return connector;
        }
    
        private static void getStatement() { 
            //驱动程序名
            String driver = "com.mysql.cj.jdbc.Driver";
            // URL指向要访问的数据库名scutcs
            String url = "jdbc:mysql://从库IP:3306/数据库名?useLegacyDatetimeCode=false&serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull";
            // MySQL配置时的用户名
            String user = "账号";
            // Java连接MySQL配置时的密码
            String password = "密码";
    
            try { 
                // 加载驱动程序
                Class.forName(driver);
    
                // 连续数据库
                Connection conn = DriverManager.getConnection(url, user, password);
                if (!conn.isClosed()) { 
                    System.out.println("Succeeded connecting to the Database!");
                }
    
                // statement用来执行SQL语句
                statement = conn.createStatement();
            } catch (ClassNotFoundException | SQLException e) { 
                e.printStackTrace();
            }
        }
    
        private static void findEntry(List<Entry> entrys) { 
            for (Entry entry : entrys) { 
                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) { 
                    continue;
                }
    
                CanalEntry.RowChange rowChage;
                try { 
                    rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) { 
                    throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                            e);
                }
    
                CanalEntry.EventType eventType = rowChage.getEventType();
                System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                        entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                        entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                        eventType));
    
                for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) { 
                    if (eventType == CanalEntry.EventType.DELETE) { 
                        String sql = "delete from " + entry.getHeader().getTableName();
                        sql += deleteSql(rowData.getBeforeColumnsList());
                        doSql(sql);
                    } else if (eventType == CanalEntry.EventType.INSERT) { 
                        String sql = "replace into " + entry.getHeader().getTableName();
                        sql += insertSql(rowData.getAfterColumnsList());
                        doSql(sql);
                    } else { 
                        String sql = "update " + entry.getHeader().getTableName();
                        sql += updateSql(rowData.getAfterColumnsList());
                        doSql(sql);
                    }
                }
            }
        }
    
        private static String updateSql(List<Column> columns) { 
            StringBuilder str = new StringBuilder(" set ");
    
            CanalField canalField = getField(columns);
            List<String> fields = canalField.getFields();
            List<String> values = canalField.getValues();
    
            for (int i = 0; i < fields.size(); i++) { 
                str.append(fields.get(i)).append(" = ").append(values.get(i)).append(",");
            }
    
            str.deleteCharAt(str.length() - 1);
            //以主键为条件更新
            str.append(" where id = ").append(canalField.getId());
            return str.toString();
        }
    
        private static String insertSql(List<Column> columns) { 
            StringBuilder str = new StringBuilder(" ( ");
            CanalField canalField = getField(columns);
            List<String> fields = canalField.getFields();
            List<String> values = canalField.getValues();
    
            str.append(String.join(",", fields))
                    .append(") values (")
                    .append(String.join(",", values))
                    .append(" ) ");
            return str.toString();
        }
    
        private static String deleteSql(List<Column> columns) { 
            StringBuilder str = new StringBuilder(" where id = ");
            String id = "";
            for (Column column : columns) { 
                if ("id".equals(column.getName())) { 
                    id = column.getValue();
                }
            }
            str.append(id);
            return str.toString();
        }
    
        private static CanalField getField(List<Column> columns) { 
            CanalField canalField = new CanalField();
    
            List<String> field = new ArrayList<>();
            List<String> value = new ArrayList<>();
            for (Column column : columns) { 
                if (StringUtils.isEmpty(column.getValue())) { 
                    continue;
                }
    
                //获得id（主键）字段值
                if ("id".equals(column.getName())) { 
                    canalField.setId(column.getValue());
                }
    
                //获得字段和值
                int sqlType = column.getSqlType();
                if (sqlType <= 8 && sqlType >= 2 || sqlType <= -5 && sqlType >= -7) { 
                    field.add("`" + column.getName() + "`");
                    value.add(column.getValue());
                } else { 
                    field.add("`" + column.getName() + "`");
                    value.add("'" + column.getValue() + "'");
                }
            }
    
            canalField.setFields(field);
            canalField.setValues(value);
            return canalField;
        }
    
        private static void doSql(String sql) { 
            try { 
                // 要执行的SQL语句
                statement.executeUpdate(sql);
            } catch (Exception e) { 
                System.out.println("执行失败：" + sql);
                e.printStackTrace();
            }
        }
    
    }

CanalField.class

package canal;
    
    import java.util.List;
    
    /** * @Author: LailaiMonkey * @Description： * @Date：Created in 2021-01-05 14:03 * @Modified By： */
    public class CanalField { 
    
        private String id;
    
        private List<String> fields;
    
        private List<String> values;
    
        public String getId() { 
            return id;
        }
    
        public void setId(String id) { 
            this.id = id;
        }
    
        public List<String> getFields() { 
            return fields;
        }
    
        public void setFields(List<String> fields) { 
            this.fields = fields;
        }
    
        public List<String> getValues() { 
            return values;
        }
    
        public void setValues(List<String> values) { 
            this.values = values;
        }
    }

需要先初始化从库表结构及数据，启动java程序实现自己同步。—— [ \[ 源码地址 \]][_ _ 1]

[Link 1]: https://blog.csdn.net/h273979586/article/details/111590436
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNzM5Nzk1ODY_size_16_color_FFFFFF_t_70]: /images/20221119/a7485a5acf154ff293e1d7c031e1be1f.png
[_ _]: https://github.com/alibaba/canal/wiki
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNzM5Nzk1ODY_size_16_color_FFFFFF_t_70 1]: /images/20221119/729bd7f2ff564b25a7caebe1b4af88a8.png
[_ _ 1]: https://github.com/Lailaimonkey/monkey-canal