1. 首页
  2. 文章中心
  3. canal

canal 入门MySQL 准备  Canal 安装  

96 阅读 0 评论
我是靠谱客的博主 独特煎蛋，最近开发中收集的这篇文章主要介绍canal 入门MySQL 准备  Canal 安装   ，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

canal入门原理分享PPT https://download.csdn.net/download/palaomo/10311240


Canal测试环境搭建HA详细文档

机器规划

zookeeper

 node1:192.168.72.101

 node2:192.168.72.102

 node3:192.168.72.103

 canal

node1:192.168.72.101

node3: 192.168.72.103

MySQL

node1:192.168.72.101

node3: 192.168.72.103

MySQL 准备 

canal的原理是基于mysql binlog技术，所以这里一定需要开启mysql的binlog写入功能，建议配置binlog模式为row.

[mysqld]

log-bin=mysql-bin #添加这一行就ok

binlog-format=ROW #选择row模式

server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复

log_slave_updates=true  #配置多节点的MySQL是需要这个配置如果这只有一个节点的话课随意
set-variable=lower_case_table_names=1（0：大小写敏感；1：大小写不敏感）最后重启一下MySql服务即可。这个主要是针对mybatis设置的以防万一

canal的原理是模拟自己为mysql slave，所以这里一定需要做为mysql slave的相关权限.

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

 

Canal 安装  

在官网上下载最新的安装包 https://github.com/alibaba/canal/releases  我下的是canal.deployer-1.0.22.tar.gz这个版本的

 

将软件包放在 /soft 目录下

 

创建 canal 的文件目录mkdir canal22

解压tar zxvf /soft/canal.deployer-1.0.22.tar.gz   -C /soft/canal22

 

进入canal 的安装目录

cd /soft/canal22

 

修改canal的instance实例名称

mv conf/example/  conf/binlog

 

配置canal公共配置

canal配置方式有两种：

  1. ManagerCanalInstanceGenerator： 基于manager管理的配置方式，目前alibaba内部配置使用这种方式。大家可以实现CanalConfigClient，连接各自的管理系统，即可完成接入。
  2. SpringCanalInstanceGenerator：基于本地spring xml的配置方式，目前开源版本已经自带该功能所有代码，建议使用

这里我用得是第二种

目前默认支持的instance.xml有以下几种：

  1. spring/memory-instance.xml
  2. spring/file-instance.xml
  3. spring/default-instance.xml
  4. spring/group-instance.xml

 

memory-instance.xml介绍：

所有的组件(parser , sink , store)都选择了内存版模式，记录位点的都选择了memory模式，重启后又会回到初始位点进行解析

特点：速度最快，依赖最少(不需要zookeeper)

场景：一般应用在quickstart，或者是出现问题后，进行数据分析的场景，不应该将其应用于生产环境

 

file-instance.xml介绍：

所有的组件(parser , sink , store)都选择了基于file持久化模式，注意，不支持HA机制.

特点：支持单机持久化

场景：生产环境，无HA需求，简单可用.

 

default-instance.xml介绍：

所有的组件(parser , sink , store)都选择了持久化模式，目前持久化的方式主要是写入zookeeper，保证数据集群共享.

特点：支持HA

场景：生产环境，集群化部署.

 

group-instance.xml介绍：

主要针对需要进行多库合并时，可以将多个物理instance合并为一个逻辑instance，提供客户端访问。

场景：分库业务。比如产品数据拆分了4个库，每个库会有一个instance，如果不用group，业务上要消费数据时，需要启动4个客户端，分别链接4个instance实例。使用group后，可以在canal server上合并为一个逻辑instance，只需要启动1个客户端，链接这个逻辑instance即可.

 

修改公共配置文件

vi conf/canal.properties

 

canal.destinations= binlog  #这个属性的值保证和尚敏修改的instance实例名称一样即可

#canal.instance.global.manager.address =127.0.0.1:1099

#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml

#canal.instance.global.spring.xml =classpath:spring/file-instance.xml 

canal.instance.global.spring.xml =classpath:spring/default-instance.xml

canal.zkServers=192.168.9.3:2181,192.168.9.4:2181,192.168.9.5:2181#增加上zookeeper集群的连接

上面这个是一个完整的HA的配置

 

配置instance 实例

编辑实例的配置文件

vi conf/binlog/instance.properties

# position info 需要改成自己的数据库信息
canal.instance.master.address = 192.168.72.101:3306  
canal.instance.master.journal.name = 
canal.instance.master.position = 
canal.instance.master.timestamp = 
#这一段是MySQL从节点的配置和主节点类似
canal.instance.standby.address = 192.168.72.103:3306 canal.instance.standby.journal.name =
canal.instance.standby.position = 
canal.instance.standby.timestamp =
 
## detecing config
canal.instance.detecting.enable = true ## 需要开启心跳检查
canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now() ##心跳检查sql，也可以选择类似select 1的query语句
canal.instance.detecting.interval.time = 3 ##心跳检查频率
canal.instance.detecting.retry.threshold = 3  ## 心跳检查失败次数阀值，超过该阀值后会触发mysql链接切换，比如切换到standby机器上继续消费binlog
canal.instance.detecting.heartbeatHaEnable = true ## 心跳检查超过失败次数阀值后，是否开启master/standby的切换. 
# table regex
canal.instance.filter.regex = pgls_dev\..*
# table black regex
canal.instance.filter.black.regex =
#################################################
 

这是一个完整的instance实例文件配置

就目前的版本来说最多支持一个standby

 

 

将改好的配置分发到其他机器

 

scp -r canal22/ root@192.168.72.103:/soft/

 

在两台机器上分别启动canal、

sh bin/startup.sh

查看启动日志

tail -100 logs/binlog/binlog.log

正常情况下应该只有一台机器正常启动并且有日志产生，两台机器会都有CanalLauncher 这个进程的如果有就说明HA没有问题

测试

 

启动zookeeper客户端

我的虚拟机是配过环境变量的 zkCli.sh start

1.  # 获取正在运行的canal server  

2.  get /otter/canal/destinations/binlog/running  

3.  # 获取正在连接的canal client  

4.  get /otter/canal/destinations/binlog /1001/running  

5.  # 获取当前最后一次消费车成功的binlog  

6.  get /otter/canal/destinations/binlog /1001/cursor  

 

验证高可用测试

 

首先验证canal服务端是否能够自动切换和接管

在一个zookeeper集群中启动两个canal服务端比如说是node1，node3，正常的情况是先在zookeeper中注册的那个进程成为active的我测试的是node1是先注册的，那node3上就是standby的状态，但是两台机器（node1，node3）上面都应该有CanalLauncher这个进程

然后手动停掉node1上面的CanalLauncher进程使用kill也行正常stop也行，我用是stop.sh现在node3上的服务应该正常接管（就是去看看启动日志是否正常），然后将node1的服务启动查看应该是standby的状态，如果是standby的状态不会有新日志的产生，然后将node3上的CanalLauncher进程停掉看看node1上的服务有没有正常的打印启动日志

 

验证canal 客户端程序是高可用的

同时启动两个canal消费端程序，此时应该只有一个能正常运行另一个应该是standby的状态，一直在心跳检查无询问zookeeper当前活动的进程是否是活动的，停掉活动的这个进程，

standby的会自动切换成 active 的将停掉的进程再次启动查看是否是standby的状态，然后将当前活动的进程停掉，验证另一个是否会接管

验证MySQL 集群模式的高可用（已验证通过）

设定 node1 master ，node3 slave

同时启动两台node1，node3 mysql服务canal 服务端和消费端均应该是正能常使用  停掉node1 mysql服务  canal服务端回去取slave 节点node3的binlog，此时 canal 服务端和消费端均应该是正能常使用 。重新启动停掉的node1上mysql服务canal 服务端和消费端均应该是正能常使用，然后关闭slave（node3）服务，验证canal服务端会重新去活动的节点node1上取binlog ，canal 服务端和消费端均应该是正能常使用而且验证是否回滚消费过的数据

 

mysql单节点数据完整性测试

验证方法：

 启动服务之前记录kafka页面上的消息数量 

 启动服务查看kafkaweb 页面上消息的数量变化

 关闭mysql服务 canal服务端和消费端午明显变化，程序正常

 重新启动mysql服务 canal没有重新消费binlog

并且新生成的binlog能够正常消费

查看canal 消费程序的日志用sum表中的主键在日志中搜索，未发现发送了多次相同的记录

 

 

 

 

 

 

 

 

 

 

下面是一个测试代码需要修改zookeeper 的连接地址和instance实例名，不用使用用户和密码，直接通过获取IP的方式即可连接

 

1.  /**

2.   * Created by hc .

3.   */ 

4.  import java.net.InetSocketAddress; 

5.  import java.util.List; 

6.    

7.  import com.alibaba.otter.canal.client.CanalConnector; 

8.  import com.alibaba.otter.canal.common.utils.AddressUtils; 

9.  import com.alibaba.otter.canal.protocol.Message; 

10. import com.alibaba.otter.canal.protocol.CanalEntry.Column; 

11. import com.alibaba.otter.canal.protocol.CanalEntry.Entry; 

12. import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; 

13. import com.alibaba.otter.canal.protocol.CanalEntry.EventType; 

14. import com.alibaba.otter.canal.protocol.CanalEntry.RowChange; 

15. import com.alibaba.otter.canal.protocol.CanalEntry.RowData; 

16. import com.alibaba.otter.canal.protocol.exception.*; 

17.   

18.   

19. import com.alibaba.otter.canal.client.*; 

20. import com.alibaba.otter.canal.client.impl.running.ClientRunningMonitor; 

21. import org.jetbrains.annotations.NotNull; 

22.   

23. publicclassClientSample{ 

24.   

25.     publicstaticvoid main(String args[]){ 

26.         // 创建链接 

27.         CanalConnector connector =CanalConnectors.newClusterConnector("192.168.213.44:4180,192.168.213.45:4180,192.168.213.46:4180","example","",""); 

28.   

29.         int batchSize =1; 

30.         int emptyCount =0; 

31.         while(true){ 

32.             try{ 

33.                 connector.connect(); 

34.                 connector.subscribe(".*\..*"); 

35.                 while(true){ 

36.                     Message messages = connector.getWithoutAck(1000); 

37.                     long bachId = messages.getId(); 

38.                     int size = messages.getEntries().size(); 

39.                     if(bachId ==-1|| size ==0){ 

40.                         try{ 

41.                             Thread.sleep(1000); 

42.                         }catch(InterruptedException e){ 

43.                             e.printStackTrace(); 

44.                         } 

45.                         System.out.println("NoDATA!!!!!!!!!!!!!!!!!!!!!!!!"); 

46.                     }else{ 

47.                         printEntry(messages.getEntries()); 

48.                     } 

49.                 } 

50.             }catch(Exception e){ 

51.                 System.out.println("============================================================connectcrash"); 

52.             }finally{ 

53.                 connector.disconnect(); 

54.             } 

55.         } 

56.     } 

57.   

58.     privatestaticvoid printEntry(@NotNullList<Entry> entrys){ 

59.         for(Entry entry : entrys){ 

60.             if(entry.getEntryType()==EntryType.TRANSACTIONBEGIN|| entry.getEntryType()==EntryType.TRANSACTIONEND){ 

61.                 continue; 

62.             } 

63.   

64.             RowChange rowChage =null; 

65.             try{ 

66.                 rowChage =RowChange.parseFrom(entry.getStoreValue()); 

67.             }catch(Exception e){ 

68.                 thrownewRuntimeException("ERROR ##parser of eromanga-event has an error , data:"+ entry.toString(), 

69.                         e); 

70.             } 

71.   

72.             EventType eventType = rowChage.getEventType(); 

73.             System.out.println(String.format("================>binlog[%s:%s] , name[%s,%s] , eventType : %s", 

74.                     entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(), 

75.                     entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), 

76.                     eventType)); 

77.   

78.             for(RowData rowData : rowChage.getRowDatasList()){ 

79.                 if(eventType ==EventType.DELETE){ 

80.                     printColumn(rowData.getBeforeColumnsList()); 

81.                 }elseif(eventType ==EventType.INSERT){ 

82.                     printColumn(rowData.getAfterColumnsList()); 

83.                 }else{ 

84.                     System.out.println("------->before"); 

85.                     printColumn(rowData.getBeforeColumnsList()); 

86.                     System.out.println("------->after"); 

87.                     printColumn(rowData.getAfterColumnsList()); 

88.                 } 

89.             } 

90.         } 

91.     } 

92.   

93.     privatestaticvoid printColumn(@NotNullList<Column> columns){ 

94.         for(Column column : columns){ 

95.             System.out.println(column.getName()+" : "+ column.getValue()+"    update="+ column.getUpdated()); 

96.         } 

97.     } 

98. } 

 

最后

以上就是独特煎蛋为你收集整理的canal 入门MySQL 准备  Canal 安装   的全部内容，希望文章能够帮你解决canal 入门MySQL 准备  Canal 安装   所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。

本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。
点赞(64)

相关文章

canal安装和使用启动客户端ClientExample
canal安装和使用启动客户端ClientExample
canal指定binlog位点启动
canal指定binlog位点启动
canal 表字段下划线和实体类驼峰 导致属性对不上显示为空的问题
canal 表字段下划线和实体类驼峰 导致属性对不上显示为空的问题
Canal注意点
Canal注意点
canal 入门MySQL 准备  Canal 安装  
canal 入门MySQL 准备  Canal 安装  
canal分库配置group-instance.xml
canal分库配置group-instance.xml
docker启动canal后无法监控mysql日志变化
docker启动canal后无法监控mysql日志变化
Canal入门1 Canal环境准备2 Canal部署安装3 数据拉取测试
Canal入门1 Canal环境准备2 Canal部署安装3 数据拉取测试

评论列表共有 0 条评论

立即
投稿
返回
顶部