1、准备：
github:https://github.com/alibaba/canal
里面有包括canal的文档，server端 client端的 例子 源码包等等。
2、canal能干吗：
项目做集群时，各服务器内数据库数据的实时更新同步，力求数据的时效性，完整性，正确性等
数据库镜像 数据库实时备份 多级索引 (卖家和买家各自分库索引) search build 业务cache刷新 价格变化等重要业务消息
keyword：数据库同步，增量订阅&消费。
3、canal工作原理：
从上层来看，复制分成三步： 第一：master将改变记录到二进制日志(binary log)中（这些记录叫做二进制日志事件，binary log events，可以通过show binlog events进行查看）； 第二：slave将master的binary log events拷贝到它的中继日志(relay log)； 第三：slave重做中继日志中的事件，将改变反映它自己的数据。
4、部署canal：
部署canal-server：
（1）开启MySQL的binlog功能，并配置binlog模式为row。
在my.cnf 加入如下：
[html] view plain copy [mysqld] log-bin=mysql-bin #添加这一行就ok binlog-format=ROW #选择row模式 server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复
（2）在mysql中 配置canal数据库管理用户，配置相应权限（repication权限）
[sql] view plain copy CREATE USER canal IDENTIFIED BY 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ; FLUSH PRIVILEGES; （3）下载canal https://github.com/alibaba/canal/releases
解压到相应文件夹
[html] view plain copy tar -zxvf canal canal canal 文件目录结构
[html] view plain copy drwxr-xr-x 2 jianghang jianghang 136 2013-02-05 21:51 bin drwxr-xr-x 4 jianghang jianghang 160 2013-02-05 21:51 conf drwxr-xr-x 2 jianghang jianghang 1.3K 2013-02-05 21:51 lib drwxr-xr-x 2 jianghang jianghang 48 2013-02-05 21:29 logs 修改配置 instance.properties
[html] view plain copy vim canal/conf/example/instance.properties [html] view plain copy ################################################# ## mysql serverId canal.instance.mysql.slaveId = 1234 # position info，需要改成自己的数据库信息 canal.instance.master.address = 127.0.0.1:3306 canal.instance.master.journal.name = canal.instance.master.position = canal.instance.master.timestamp = #canal.instance.standby.address = #canal.instance.standby.journal.name = #canal.instance.standby.position = #canal.instance.standby.timestamp = # username/password，需要改成自己的数据库信息 canal.instance.dbUsername = canal canal.instance.dbPassword = canal canal.instance.defaultDatabaseName = canal_test canal.instance.connectionCharset = UTF-8 # table regex canal.instance.filter.regex = .*\\..* ################################################# 然后cd到bin目录 启动和停止canal-server
启动
[html] view plain copy ./startup.sh 停止
[html] view plain copy ./stop.sh 验证启动状态，查看log文件
[html] view plain copy vim canal/log/canal/canal.log [html] view plain copy 2014-07-18 10:21:08.525 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server. 2014-07-18 10:21:08.609 [main] INFO com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[10.12.109.201:11111] 2014-07-18 10:21:09.037 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ...... 上述日志信息显示启动canal成功
运行canal-client实例:
（1）建立实例maven工程
[html] view plain copy mvn archetype:create -DgroupId=com.alibaba.otter -DartifactId=canal.sample （2）添加pom依赖：
[html] view plain copy <dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.0.12</version> </dependency> （3）更新依赖 mvn install
（4）canal-client.Java 实例代码
[java] view plain copy /** * Created by hp on 14-7-17. */ import java.net.InetSocketAddress; import java.util.List; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.common.utils.AddressUtils; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.CanalEntry.Column; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; import com.alibaba.otter.canal.protocol.CanalEntry.EventType; import com.alibaba.otter.canal.protocol.CanalEntry.RowChange; import com.alibaba.otter.canal.protocol.CanalEntry.RowData; import com.alibaba.otter.canal.client.*; import org.jetbrains.annotations.NotNull; public class ClientSample { public static void main(String args[]) { // 创建链接 CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(), 11111), "example", "", ""); int batchSize = 1000; int emptyCount = 0; try { connector.connect(); connector.subscribe(".*\\..*"); connector.rollback(); int totalEmtryCount = 1200; while (emptyCount < totalEmtryCount) { Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据 long batchId = message.getId(); int size = message.getEntries().size(); if (batchId == -1 || size == 0) { emptyCount++; System.out.println("empty count : " + emptyCount); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } else { emptyCount = 0; // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size); printEntry(message.getEntries()); } connector.ack(batchId); // 提交确认 // connector.rollback(batchId); // 处理失败, 回滚数据 } System.out.println("empty too many times, exit"); } finally { connector.disconnect(); } } private static void printEntry(@NotNull List<Entry> entrys) { for (Entry entry : entrys) { if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) { continue; } RowChange rowChage = null; try { rowChage = RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e); } 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 (RowData rowData : rowChage.getRowDatasList()) { if (eventType == EventType.DELETE) { printColumn(rowData.getBeforeColumnsList()); } else if (eventType == EventType.INSERT) { printColumn(rowData.getAfterColumnsList()); } else { System.out.println("-------> before"); printColumn(rowData.getBeforeColumnsList()); System.out.println("-------> after"); printColumn(rowData.getAfterColumnsList()); } } } } private static void printColumn(@NotNull List<Column> columns) { for (Column column : columns) { System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated()); } } } （5）运行java实例
启动后看到控制端信息：
[html] view plain copy empty count : 1 empty count : 2 empty count : 3 empty count : 4 （6）触发数据库变更
[html] view plain copy create table test ( uid int (4) primary key not null auto_increment, name varchar(10) not null); insert into test (name) values('10');
（7）client 抓取mysql信息:
[html] view plain copy ================> binlog[mysql-bin.000016:3281] , name[canal_test,test] , eventType : INSERT uid : 7 update=false name : 10 update=false empty count : 1 empty count : 2 5、部署过程中产生问题：
（1）启动失败，log日志中地址正在使用
1、11111端口正在被占用 可以用 ls -i:11111 查看监听进程谁占用端口 或者 用 ps -ef | grep 11111 查看哪个进程占用端口号 然后 kill -9 进程号 杀掉占用进程
2、可以编辑 canal/conf/canal.properties 中的端口号 ，改为不占用的端口
（2）canal无法抓取mysql触发数据库改变的信息
1、检查mysql是否打开binlog写入功能 检查binlog 是否为行模式。
[html] view plain copy show variables like "binlog_format" 2、检查my.cnf 和 instance.properties 等配置文件填写信息是否正确。
3、检查client 代码 调试实例代码
4、版本兼容问题，canal 1.8 换成 canal 1.7 继续测试
5、查看所有日志文件 分析日志
