Canal的安装与使用

1、Canal介绍

　　Canal的原理就是它本身假装成slave， 向mysql发送dump协议，MySQL master接收到dump请求以后推送binlog文件给slave， 也就是canal。　　

 

2、Canal安装

　　1. 下载Canal 

　　　wget https://github.com/alibaba/canal/releases/download/canal-1.0.24/canal.deployer-1.0.24.tar.gz

　　2. 解压到/opt/softwares/canal目录， 解压完以后以下图所示：

　　3. 配置instance

 

 

　　4. 修改canal.properties

 

 

3、Mysql 安装

　　一、mysql 安装

　　　　yum install mysql

　　　　yum install mysql-server 

　　二、启动mysql

　　　　/etc/init.d/mysqld start 或者sevice mysqld start

　　三、设置root用户密码

　　　　mysqladmin -u root password '123456' 

　　四、登陆msyql

　　　　mysql -uroot -p123456

　　五、检查并开启binlog复制功能及binlog模式是否为ROW模式

　　　　参考: binlog详解

 

4、Canal抽取binlog

　　Canal只是假装成slave抽取binlog，Canal拿到binlog以后还须要交给业务方去作响应的处理，那么怎么去交给业务方呢？通常都是Canal获取到binlog以后写到kafka里，业务方订阅kafka topic消费binlog，完成业务逻辑处理。

　　可是Canal不能直接写Kafka, 因此还须要有个client链接Canal，Canal获取binlog以后交给Client， Client在往Kafka里写binlog消息，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 java.net.InetSocketAddress; import java.util.List; public class CanalClientExample { public static void main(String[] args) { CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.41.254", 11111), "example", "canal", "canal"); try { int batchSize = 1000; connector.connect(); connector.subscribe("zhengxinv6\\..*"); connector.rollback(); 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 e) { e.printStackTrace(); } continue; } System.out.println("batchId = [" + batchId + "]"); printEntry(message.getEntries()); connector.ack(batchId); //提交确认 //connector.rollback(batchId);
 } } finally { connector.disconnect(); } } private static void printEntry(List<CanalEntry.Entry> entrys) { for (CanalEntry.Entry entry : entrys) { if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) { continue; } CanalEntry.RowChange rowChange = null; try { rowChange = 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 = rowChange.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: rowChange.getRowDatasList()) { if (eventType == CanalEntry.EventType.DELETE) { printColumn(rowData.getBeforeColumnsList()); } else if (eventType == CanalEntry.EventType.INSERT) { printColumn(rowData.getAfterColumnsList()); } else { System.out.println("-------> before"); printColumn(rowData.getBeforeColumnsList()); System.out.println("-------> after"); printColumn(rowData.getAfterColumnsList()); } } } } private static void printColumn(List<CanalEntry.Column> columns) { for (CanalEntry.Column column : columns) { System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated()); } } }

 

 

5、Canal使用过程出现的问题及解决方法

　　参考：canal报错解决方法

　　

 

 

参考：https://www.jianshu.com/p/6299048fad66