1.canal简介
canal可以用来监控数据库数据的变化，从而获得新增、修改的数据。
原理相对比较简单：
（1）canal模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议
（2） mysql master收到dump请求，开始推送binary log给slave(也就是canal)
（3） canal解析binary log对象(原始为byte流)
2.环境部署
（1）mysql开启binlog模式
查看当前mysql是否开启binlog模式，如果log_bin的值为OFF是未开启，为ON是已开启。
SHOW VARIABLES LIKE '%log_bin%'
修改/etc/my.cnf 需要开启binlog模式。（修改完成之后，重启mysqld的服务。）
log-bin=mysql-bin
binlog-format=ROW
server_id=1
进入mysql
mysql -h localhost -u root -p
创建账号 用于测试使用（使用root账号创建用户并授予权限）
create user canal@'%' IDENTIFIED by 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;
3.canal服务端安装配置
（1）下载地址canal
https://github.com/alibaba/canal/releases/tag/canal-1.0.24
（2）下载之后 上传到linux系统中，解压缩到指定的目录/usr/local/canal
解压缩之后的目录结构如下：
（3）修改 exmaple下的实例配置（修改如图所示的几个参数）
vi conf/example/instance.properties
（4）指定读取位置
进入mysql中执行下面语句查看binlog所在位置
如果fifile中binlog文件不为 mysql-bin.000001 可以重置mysql
mysql> reset master;
查看canal配置文件
vim /usr/local/canal/conf/example/meta.dat
找到对应的binlog信息更改一致即可
"journalName":"mysql-bin.000001","position":120,"
注意：如果不一致，可能导致以下错误
c.a.otter.canal.server.netty.handler.SessionHandler - something goes wrong with channel:[id: 0x7f2e9be3, /192.168.200.56:52225 => /192.168.200.128:11111],exception=java.io.IOException: Connection reset by peer
（5）启动服务：
[root@localhost canal]# ./bin/startup.sh
（6）查看日志：
cat /usr/local/canal/logs/canal/canal.log
4.数据监控微服务
当用户执行数据库的操作的时候，binlog日志会被canal捕获到，并解析出数据。我们就可以将解析出来的数据进行相应的逻辑处理。
https://github.com/chenqian56131/spring-boot-starter-canal
以上开源项目，实现了springboot与canal的集成。比原生的canal更加优雅。
使用前需要将starter-canal安装到本地仓库。我们可以参照它提供的canal-test，进行代码实现。
（1）创建工程模块changgou_canal，pom引入依赖
<dependency>
<groupId>com.xpand</groupId>
<artifactId>starter-canal</artifactId>
<version>0.0.1-SNAPSHOT</version>
</dependency>
（2）创建包com.changgou.canal ，包下创建启动类
@SpringBootApplication
@EnableCanalClient //声明当前的服务是canal的客户端
public class CanalApplication {
public static void main(String[] args) {
SpringApplication.run(CanalApplication.class,args);
}
}
（3）添加配置文件application.properties
canal.client.instances.example.port=11111
canal.client.instances.example.batchSize=1000
spring.rabbitmq.host=192.168.200.128
（4）创建com.changgou.canal.listener包，包下创建类
@CanalEventListener //声明当前的类是canal的监听类
public class BusinessListener {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
*
* @param eventType 当前操作数据库的类型
* @param rowData 当前操作数据库的数据
*/
@ListenPoint(schema = "business",table = "tb_ad")
public void adUpdate(CanalEntry.EventType eventType,CanalEntry.RowData rowData){
System.out.println("广告表数据发生改变");
//获取改变之前的数据
rowData.getBeforeColumnsList().forEach((c)-> System.out.println("改变前的数据:"+c.getName()+"::"+c.getValue()));
//获取改变之后的数据
rowData.getAfterColumnsList().forEach((c)-> System.out.println("改变之后的
数据:"+c.getName()+"::"+c.getValue()));
}
}
测试：启动数据监控微服务，修改business的tb_ad表，观察控制台输出。
