git地址:阿里巴巴Canal的Git地址
Canal基于日志增量订阅和消费的业务包括:
- 数据库镜像、数据库实时备份
- 索引构建和实时维护(拆分异构索引、倒排索引)
- 业务cache刷新、带业务逻辑的增量数据处理
Mysql 的数据同步的架构图:
![在这里插入图片描述](https://img-blog.csdnimg.cn/3462763200d8469cb71cab519c1f152c.png)
Canal是把自己伪装成一个从库:
![在这里插入图片描述](https://img-blog.csdnimg.cn/3d94f3dbb1674e60802eae9ef53e2d33.png)
Canal的优点: 实时性好、分布式、ACK机制
Canal的缺点:
- 只支持增量同步,不支持全量同步
- MYSQL -> ES、RDB
- 一个instance只能有一个消费端消费
- 单点压力大
Canal的组件:
![在这里插入图片描述](https://img-blog.csdnimg.cn/9085539c740d4db1b56f5e42635462c9.png)
Cancal 和 ES 的数据同步的架构: Canal可以和Kafka无缝连接
![在这里插入图片描述](https://img-blog.csdnimg.cn/8dfc361297a84ca395927d7ee8de0a98.png)
Cancal配置:
- 开启mysql得bin-log日志
vim /etc/my.cnf
[mysqld]
log-bin=mysql-bin
binlog-format=ROW
server_id=1
mysql -uroot -proot
show variables like '%log_bin%';
CREATE USER root IDENTIFIED BY 'root';
GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' IDENTIFIED BY 'root' WITH GRANT OPTION;
-- CREATE USER canal IDENTIFIED BY 'canal';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal' WITH GRANT OPTION;
-- GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;
- Canal 配置
mkdir /usr/local/canal
tar -zxvf canal.deployer-1.1.4.tar.gz -C /usr/local/canal/
vim /usr/local/canal/conf/canal.properties
canal.port = 11111
vim /usr/local/canal/conf/example/instance.properties
canal.instance.mysql.slaveId=1001
canal.instance.master.address=192.168.11.31:3306
canal.instance.dbUsername=root
canal.instance.dbPassword=root
canal.instance.connectionCharset = UTF-8
cd /usr/local/canal/bin
./startup.sh
cat /usr/local/canal/logs/canal/canal.log
tail -f -n 100 /usr/local/canal/logs/example/example.log
Canal 与 MQ 整合:
官方文档: Canal Kafka RocketMQ QuickStart
Java 操作 Canal例子:
官方java例子: ClientSample.java
package com.alibaba.otter.canal.sample;
import java.net.InetSocketAddress;
import java.util.List;
import com.alibaba.otter.canal.client.CanalConnectors;
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;
public class SimpleCanalClientExample {
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 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;
printEntry(message.getEntries());
}
connector.ack(batchId);
}
System.out.println("empty too many times, exit");
} finally {
connector.disconnect();
}
}
private static void printEntry(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(List<Column> columns) {
for (Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
}
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)