mysql本身是支持主从的(master slave)，原理就是master产生的binlog日志记录了所有的增删改语句，将binlog发送到slave节点进行执行即可完成数据的同步。

canal是阿里开源的一个中间件，它就是通过解析binlog来完成数据变更的监听的。

https://github.com/alibaba/canal

可以看到，canal是这样工作的：canal有一个server工程，该server自己伪装为mysql的一个slave节点，然后向master请求所有的变更日志。同时canal有一个client工程，通过添加client的sdk，我们就可以在项目里监听到server端传来的数据变更信息，从而达到监听数据变化的目的。

安装配置canal

在这里https://github.com/alibaba/canal/releases，下载canal releases包，如https://github.com/alibaba/canal/releases/download/canal-1.0.22/canal.deployer-1.0.22.tar.gz。

这个就是server端，下载后解压，如图

打开conf/example下的instance.properties文件，编辑#################################################

## 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 = test

canal.instance.connectionCharset = UTF-8

# table regex

canal.instance.filter.regex = .*\\..*

# table black regex

canal.instance.filter.black.regex =

#################################################

address设置为mysql的连接地址，defaultDatabaseName设置为自己要监听的库名，如test。

在mysql命令行，创建一个新用户，作为slaveCREATE USER canal IDENTIFIED BY 'canal';

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';

-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;

FLUSH PRIVILEGES;

对应配置文件里的canal用户。到此配置完毕。

执行bin目录下的startup.sh

启动后可以在logs目录下查看日志。在example目录下的example.log，如果没有报错，说明启动成功。

canal客户端编写

服务端启动完毕后，在客户端即可监听test库的变化。

新建一个java maven项目，pom.xml里添加依赖

com.alibaba.otter

canal.client

1.0.12

package com.mindata;

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

import com.alibaba.otter.canal.client.CanalConnectors;

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

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

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

import java.net.InetSocketAddress;

import java.util.List;

/**

* A Camel Application

*/

public class MainApp {

/**

* A main() so we can easily run these routing rules in our IDE

*/

public static void main(String... args) throws Exception {

// 创建链接

CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),

11111), "example", "", "");

int batchSize = 1000;

int emptyCount = 0;

try {

connector.connect();

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

connector.rollback();

int totalEmptyCount = 120;

while (emptyCount < totalEmptyCount) {

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) {

}

} 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(List entrys) {

for (CanalEntry.Entry entry : entrys) {

if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry

.EntryType

.TRANSACTIONEND) {

continue;

}

CanalEntry.RowChange rowChage = null;

try {

rowChage = 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 = 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 (CanalEntry.RowData rowData : rowChage.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 columns) {

for (CanalEntry.Column column : columns) {

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

}

}

}

注意newSingleConnector那里，设置canal server的ip和端口，端口默认为11111。example是和conf目录下的相对应。

如果你在这里新建个目录abc，目录里也放一个instance.properties，那么就可以在客户端监听abc。

启动客户端，查看控制台输出。

启动后，就可以打印empty count，此时你可以操作数据库里的test库，做任何增删改的操作，然后就能在控制台看到打印的语句。

可以看到，修改的是test库里的History表，before代表修改前，id=3，calltime字段值为3，after修改后，calltime为22.customer列我没有修改，所以值为242没变。

这样就获得了对应的库里，某个表的任何一列的变化的解析。