经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 数据库/运维 » MySQL » 查看文章
canal
来源：cnblogs　　作者：废物大师兄　　时间：2022/1/17 11:00:49　　对本文有异议

1.  canal 简介

canal 主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费。

canal 工作原理：

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

官网地址：https://github.com/alibaba/canal

2.  canal 服务端

 （1）修改 /etc/my.cnf 文件，增加以下配置（PS：改完后别忘了重启数据库 systemctl restart mysqld ）

  1. [mysqld]
  2. log-bin=mysql-bin # 开启 binlog
  3. binlog-format=ROW # 选择 ROW 模式
  4. server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

备忘录

  1. show variables like 'log_bin';
  2. show binary logs;
  3. show master status;
  4. select * from mysql.`user`;

（2）授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant

  1. CREATE USER canal IDENTIFIED BY 'Canal@123456';  
  2. GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
  3. -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
  4. FLUSH PRIVILEGES;

 （3）下载canal，并解压缩

https://github.com/alibaba/canal/releases

（4）修改配置

  1. vim conf/example/instance.properties

启动

  1. sh bin/startup.sh

查看日志

  1. # 查看 server 日志
  2. vi logs/canal/canal.log
  4. # 查看 instance 的日志
  5. vi logs/example/example.log
  7. # 关闭
  8. sh bin/stop.sh

3. canal 客户端

依赖

  1. <dependency>
  2.     <groupId>com.alibaba.otter</groupId>
  3.     <artifactId>canal.client</artifactId>
  4.     <version>1.1.5</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>com.alibaba.otter</groupId>
  8.     <artifactId>canal.protocol</artifactId>
  9.     <version>1.1.5</version>
  10. </dependency>

配置

  1. canal.server.host=192.168.28.32
  2. canal.server.port=11111
  3. canal.server.destination=example
  4. canal.server.username=canal
  5. canal.server.password=canal
  7. spring.kafka.bootstrap-servers=192.168.28.32:9092

logback.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <configuration scan="true" scanPeriod="30 seconds" debug="false">
  3.     <property name="log.charset" value="utf-8" />
  4.     <property name="log.pattern" value="%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n" />
  5.     <property name="log.dir" value="./logs" />
  7.     <!--输出到控制台-->
  8.     <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
  9.         <encoder>
  10.             <pattern>${log.pattern}</pattern>
  11.             <charset>${log.charset}</charset>
  12.         </encoder>
  13.     </appender>
  14.     <appender name="file" class="ch.qos.logback.core.rolling.RollingFileAppender">
  15.         <file>${log.dir}/my-canal-client.log</file>
  16.         <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
  17.             <fileNamePattern>${log.dir}/my-canal-client.%d{yyyy-MM-dd}.log</fileNamePattern>
  18.             <maxHistory>30</maxHistory>
  19.             <totalSizeCap>3GB</totalSizeCap>
  20.         </rollingPolicy>
  21.         <encoder>
  22.             <pattern>${log.pattern}</pattern>
  23.         </encoder>
  24.     </appender>
  26.     <root level="info">
  27.         <appender-ref ref="console" />
  28.         <appender-ref ref="file" />
  29.     </root>
  30. </configuration>

测试代码

  1. package com.my.component.canal;
  3. import com.alibaba.otter.canal.client.CanalConnector;
  4. import com.alibaba.otter.canal.protocol.CanalEntry;
  5. import com.alibaba.otter.canal.protocol.Message;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.springframework.util.Assert;
  9. import java.util.List;
  11. /**
  12.  * @Author ChengJianSheng
  13.  * @Date 2022/1/13
  14.  */
  15. @Slf4j
  16. public class Demo {
  18.     private volatile boolean running = false;
  19.     private CanalConnector connector;
  20.     private Thread thread = null;
  22.     public Demo(CanalConnector connector) {
  23.         this.connector = connector;
  24.     }
  26.     public void start() {
  27.         Assert.notNull(connector, "connector is null");
  28.         thread = new Thread(this::process);
  29.         running = true;
  30.         thread.start();
  31.     }
  33.     public void stop() {
  34.         if (!running) {
  35.             return;
  36.         }
  37.         running = false;
  38.         if (thread != null) {
  39.             try {
  40.                 thread.join();
  41.             } catch (InterruptedException e) {
  42.                 // ignore
  43.             }
  44.         }
  45.     }
  47.     private void process() {
  48.         int batchSize = 5 * 1024;
  49.         while (running) {
  50.             try {
  51.                 connector.connect();
  52.                 connector.subscribe(".*\\..*");
  53.                 while (running) {
  54.                     Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
  55.                     long batchId = message.getId();
  56.                     int size = message.getEntries().size();
  57.                     if (batchId == -1 || size == 0) {
  58.                         // try {
  59.                         // Thread.sleep(1000);
  60.                         // } catch (InterruptedException e) {
  61.                         // }
  62.                     } else {
  63.                         printEntry(message.getEntries());
  64.                     }
  66.                     if (batchId != -1) {
  67.                         connector.ack(batchId); // 提交确认
  68.                     }
  69.                 }
  70.             } catch (Throwable e) {
  71.                 log.error("process error!", e);
  72.                 try {
  73.                     Thread.sleep(1000L);
  74.                 } catch (InterruptedException e1) {
  75.                     // ignore
  76.                 }
  78.                 connector.rollback(); // 处理失败, 回滚数据
  79.             } finally {
  80.                 connector.disconnect();
  81.             }
  82.         }
  83.     }
  85.     private void printEntry(List<CanalEntry.Entry> entrys) {
  86.         for (CanalEntry.Entry entry : entrys) {
  87.             if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
  88.                 continue;
  89.             }
  91.             CanalEntry.RowChange rowChage = null;
  92.             try {
  93.                 rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
  94.             } catch (Exception e) {
  95.                 throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
  96.             }
  98.             CanalEntry.EventType eventType = rowChage.getEventType();
  99.             System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
  100.                     entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
  101.                     entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
  102.                     eventType));
  104.             for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
  105.                 if (eventType == CanalEntry.EventType.DELETE) {
  106.                     printColumn(rowData.getBeforeColumnsList());
  107.                 } else if (eventType == CanalEntry.EventType.INSERT) {
  108.                     printColumn(rowData.getAfterColumnsList());
  109.                 } else {
  110.                     System.out.println("-------> before");
  111.                     printColumn(rowData.getBeforeColumnsList());
  112.                     System.out.println("-------> after");
  113.                     printColumn(rowData.getAfterColumnsList());
  114.                 }
  115.             }
  116.         }
  117.     }
  119.     private void printColumn(List<CanalEntry.Column> columns) {
  120.         for (CanalEntry.Column column : columns) {
  121.             System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
  122.         }
  123.     }
  124. }

启动

  1. package com.my.component.canal;
  3. import com.alibaba.otter.canal.client.CanalConnector;
  4. import com.alibaba.otter.canal.client.CanalConnectors;
  5. import com.alibaba.otter.canal.common.utils.AddressUtils;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.boot.CommandLineRunner;
  9. import org.springframework.kafka.core.KafkaTemplate;
  10. import org.springframework.stereotype.Component;
  12. import java.net.InetSocketAddress;
  13. import java.net.SocketAddress;
  15. /**
  16.  * @Author ChengJianSheng
  17.  * @Date 2022/1/4
  18.  */
  19. @Slf4j
  20. @Component
  21. public class MyCommandLineRunner implements CommandLineRunner {
  23.     @Autowired
  24.     private CanalServerProperties canalServerProperties;
  25.     @Autowired
  26.     private KafkaTemplate<String, String> kafkaTemplate;
  28.     @Override
  29.     public void run(String... args) throws Exception {
  30.         String destination = "example";
  31. //        String ip = AddressUtils.getHostIp();
  32. //        String ip = canalServerProperties.getHost();
  33.         String ip = "192.168.28.32";
  34.         SocketAddress address = new InetSocketAddress(ip, 11111);
  35.         CanalConnector connector = CanalConnectors.newSingleConnector(address, destination, "canal", "canal");
  37.         Demo demo = new Demo(connector);
  38.         demo.start();
  40.         Runtime.getRuntime().addShutdownHook(new Thread(() -> {
  41.             try {
  42.                 log.info("## stop the canal client");
  43.                 demo.stop();
  44.             } catch (Throwable e) {
  45.                 log.warn("##something goes wrong when stopping canal:", e);
  46.             } finally {
  47.                 log.info("## canal client is down.");
  48.             }
  49.         }));
  50.     }
  51. }

 

原文链接：http://www.cnblogs.com/cjsblog/p/15799448.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

MySQL热门文章

  • 大项目之网上书城（九）——订单Demo
  • CentOS7下MySQL服务启动失败原因及解决方法
  • MySQL从删库到跑路_高级（六）——索引
  • 55 道MySQL基础题
  • MySQL与PHP的基础与应用专题之自连接
  • 树形结构的菜单表设计与查询
  • MYSQL语句
  • 《高性能MySQL（第3版）》PDF免费下载
  • 《MySQL必知必会》知识汇总二
  • 【必知必会的MySQL知识】④DCL语言

    • MySQL推荐文章

  • 如何让SQL Server像MySQL一样拥有慢查询日志（Slow Query Log慢日志）
  • MySQL中为什么要使用索引合并(Index Merge)？
  • MySQL派生表合并优化的原理和实现
  • 使用EF 连接 数据库 SQLserver、MySql 实现 CodeFirst
  • 升级到 MySQL 8.4，MySQL 启动报错：io_setup() failed with EAGAIN
  • TiKV 源码分析之 PointGet
  • 异构数据源同步之数据同步 → DataX 使用细节
  • 一文搞懂 MySQL 日志
  • 浅析MySQL 8.0直方图原理
  • 分布式任务调度内的 MySQL 分页查询优化

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号