Canal的介绍Canal的历史由来在早期的时候,阿里巴巴公司因为杭州和美国两个地方的机房都部署了数据库实例,但因为跨机房同步数据的业务需求 ,便孕育而生出了Canal,主要是基于trigger(触发器)的方式获取增量变更。从2010年开始,阿里巴巴公司开始逐步尝试数据库日志解析,获取增量变更的数据进行同步,由此衍生出了增量订阅和消费业务。 Canal的应用场景目前普遍基于日志增量订阅和消费的业务,主要包括:
Canal的工作原理在介绍Canal的原理之前,我们先来了解下MySQL主从复制的原理。 MySQL主从复制原理
Canal工作原理
Canal的Docker环境准备因为目前容器化技术的火热,本文通过使用Docker来快速搭建开发环境,而传统方式的环境搭建,在我们学会了Docker容器环境搭建后,也能自行依葫芦画瓢搭建成功。由于本篇主要讲解Canal,所以关于Docker的内容不会涉及太多,主要会介绍Docker的基本概念和命令使用。如果你想和更多容器技术专家交流,可以加我微信liyingjiese,备注『加群』。群里每周都有全球各大公司的最佳实践以及行业最新动态。 什么是Docker
相信绝大多数人都使用过虚拟机VMware,在使用VMware进行环境搭建的时候,只需提供了一个普通的系统镜像并成功安装,剩下的软件环境与应用配置还是如我们在本机操作一样在虚拟机里也操作一遍,而且VMware占用宿主机的资源较多,容易造成宿主机卡顿,而且系统镜像本身也占用过多空间。
Docker的网络介绍Docker的网络类型有三种:
docker network create --subnet=172.18.0.0/16 mynetwork
搭建Canal环境
附上Docker的下载安装地址==> Docker Download。
##生成mysql容器 docker run -d --name mysql --net mynetwork --ip 172.18.0.6 -p 3306:3306 -e MYSQL_ROOT_PASSWORD=root mysql ##生成canal-server容器 docker run -d --name canal-server --net mynetwork --ip 172.18.0.4 -p 11111:11111 canal/canal-server ## 命令介绍 --net mynetwork #使用自定义网络 --ip #指定分配ip
MySQL的配置修改
以上只是初步准备好了基础的环境,但是怎么让Canal伪装成Salve并正确获取MySQL中的binary log呢? [mysqld] log-bin=mysql-bin # 开启binlog binlog-format=ROW # 选择ROW模式 server_id=1 # 配置MySQL replaction需要定义,不要和Canal的slaveId重复
mysql -uroot -proot # 创建账号 CREATE USER canal IDENTIFIED BY 'canal'; # 授予权限 GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ; # 刷新并应用 FLUSH PRIVILEGES;
show variables like 'log_bin'; show variables like 'log_bin'; show master status;
canal-server的配置修改
进入canal-server容器docker exec -it canal-server bash。
docker exec -it canal-server bash tail -100f canal-server/logs/example/example.log
拉取数据并同步保存到ElasticSearch本文的ElasticSearch也是基于Docker环境搭建,所以读者可执行如下命令: # 下载对镜像 docker pull elasticsearch:7.1.1 docker pull mobz/elasticsearch-head:5-alpine # 创建容器并运行 docker run -d --name elasticsearch --net mynetwork --ip 172.18.0.2 -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" elasticsearch:7.1.1 docker run -d --name elasticsearch-head --net mynetwork --ip 172.18.0.5 -p 9100:9100 mobz/elasticsearch-head:5-alpine
Student.javapackage com.example.canal.study.pojo; import lombok.Data; import java.io.Serializable; // @Data 用户生产getter、setter方法 @Data public class Student implements Serializable { private String id; private String name; private int age; private String sex; private String city; }
CanalConfig.javapackage com.example.canal.study.common; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import org.apache.http.HttpHost; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestHighLevelClient; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.net.InetSocketAddress; /** * @author haha */ @Configuration public class CanalConfig { // @Value 获取 application.properties配置中端内容 @Value("${canal.server.ip}") private String canalIp; @Value("${canal.server.port}") private Integer canalPort; @Value("${canal.destination}") private String destination; @Value("${elasticSearch.server.ip}") private String elasticSearchIp; @Value("${elasticSearch.server.port}") private Integer elasticSearchPort; @Value("${zookeeper.server.ip}") private String zkServerIp; // 获取简单canal-server连接 @Bean public CanalConnector canalSimpleConnector() { CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalIp, canalPort), destination, "", ""); return canalConnector; } // 通过连接zookeeper获取canal-server连接 @Bean public CanalConnector canalHaConnector() { CanalConnector canalConnector = CanalConnectors.newClusterConnector(zkServerIp, destination, "", ""); return canalConnector; } // elasticsearch 7.x客户端 @Bean public RestHighLevelClient restHighLevelClient() { RestHighLevelClient client = new RestHighLevelClient( RestClient.builder(new HttpHost(elasticSearchIp, elasticSearchPort)) ); return client; } }
CanalDataParser.java由于这个类的代码较多,文中则摘出其中比较重要的部分,其它部分代码可从GitHub上获取: public static class TwoTuple<A, B> { public final A eventType; public final B columnMap; public TwoTuple(A a, B b) { eventType = a; columnMap = b; } } public static List<TwoTuple<EventType, Map>> printEntry(List<Entry> entrys) { List<TwoTuple<EventType, Map>> rows = new ArrayList<>(); for (Entry entry : entrys) { // binlog event的事件事件 long executeTime = entry.getHeader().getExecuteTime(); // 当前应用获取到该binlog锁延迟的时间 long delayTime = System.currentTimeMillis() - executeTime; Date date = new Date(entry.getHeader().getExecuteTime()); SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); // 当前的entry(binary log event)的条目类型属于事务 if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) { if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN) { TransactionBegin begin = null; try { begin = TransactionBegin.parseFrom(entry.getStoreValue()); } catch (InvalidProtocolBufferException e) { throw new RuntimeException("parse event has an error , data:" + entry.toString(), e); } // 打印事务头信息,执行的线程id,事务耗时 logger.info(transaction_format, new Object[]{entry.getHeader().getLogfileName(), String.valueOf(entry.getHeader().getLogfileOffset()), String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date), entry.getHeader().getGtid(), String.valueOf(delayTime)}); logger.info(" BEGIN ----> Thread id: {}", begin.getThreadId()); printXAInfo(begin.getPropsList()); } else if (entry.getEntryType() == EntryType.TRANSACTIONEND) { TransactionEnd end = null; try { end = TransactionEnd.parseFrom(entry.getStoreValue()); } catch (InvalidProtocolBufferException e) { throw new RuntimeException("parse event has an error , data:" + entry.toString(), e); } // 打印事务提交信息,事务id logger.info("----------------\n"); logger.info(" END ----> transaction id: {}", end.getTransactionId()); printXAInfo(end.getPropsList()); logger.info(transaction_format, new Object[]{entry.getHeader().getLogfileName(), String.valueOf(entry.getHeader().getLogfileOffset()), String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date), entry.getHeader().getGtid(), String.valueOf(delayTime)}); } continue; } // 当前entry(binary log event)的条目类型属于原始数据 if (entry.getEntryType() == EntryType.ROWDATA) { RowChange rowChage = null; try { // 获取储存的内容 rowChage = RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException("parse event has an error , data:" + entry.toString(), e); } // 获取当前内容的事件类型 EventType eventType = rowChage.getEventType(); logger.info(row_format, new Object[]{entry.getHeader().getLogfileName(), String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType, String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date), entry.getHeader().getGtid(), String.valueOf(delayTime)}); // 事件类型是query或数据定义语言DDL直接打印sql语句,跳出继续下一次循环 if (eventType == EventType.QUERY || rowChage.getIsDdl()) { logger.info(" sql ----> " + rowChage.getSql() + SEP); continue; } printXAInfo(rowChage.getPropsList()); // 循环当前内容条目的具体数据 for (RowData rowData : rowChage.getRowDatasList()) { List<CanalEntry.Column> columns; // 事件类型是delete返回删除前的列内容,否则返回改变后列的内容 if (eventType == CanalEntry.EventType.DELETE) { columns = rowData.getBeforeColumnsList(); } else { columns = rowData.getAfterColumnsList(); } HashMap<String, Object> map = new HashMap<>(16); // 循环把列的name与value放入map中 for (Column column: columns){ map.put(column.getName(), column.getValue()); } rows.add(new TwoTuple<>(eventType, map)); } } } return rows; }
ElasticUtils.javapackage com.example.canal.study.common; import com.alibaba.fastjson.JSON; import com.example.canal.study.pojo.Student; import lombok.extern.slf4j.Slf4j; import org.elasticsearch.client.RestHighLevelClient; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.elasticsearch.action.DocWriteRequest; import org.elasticsearch.action.delete.DeleteRequest; import org.elasticsearch.action.delete.DeleteResponse; import org.elasticsearch.action.get.GetRequest; import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.update.UpdateRequest; import org.elasticsearch.action.update.UpdateResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.common.xcontent.XContentType; import java.io.IOException; import java.util.Map; /** * @author haha */ @Slf4j @Component public class ElasticUtils { @Autowired private RestHighLevelClient restHighLevelClient; /** * 新增 * @param student * @param index 索引 */ public void saveEs(Student student, String index) { IndexRequest indexRequest = new IndexRequest(index) .id(student.getId()) .source(JSON.toJSONString(student), XContentType.JSON) .opType(DocWriteRequest.OpType.CREATE); try { IndexResponse response = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT); log.info("保存数据至ElasticSearch成功:{}", response.getId()); } catch (IOException e) { log.error("保存数据至elasticSearch失败: {}", e); } } /** * 查看 * @param index 索引 * @param id _id * @throws IOException */ public void getEs(String index, String id) throws IOException { GetRequest getRequest = new GetRequest(index, id); GetResponse response = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT); Map<String, Object> fields = response.getSource(); for (Map.Entry<String, Object> entry : fields.entrySet()) { System.out.println(entry.getKey() + ":" + entry.getValue()); } } /** * 更新 * @param student * @param index 索引 * @throws IOException */ public void updateEs(Student student, String index) throws IOException { UpdateRequest updateRequest = new UpdateRequest(index, student.getId()); updateRequest.upsert(JSON.toJSONString(student), XContentType.JSON); UpdateResponse response = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT); log.info("更新数据至ElasticSearch成功:{}", response.getId()); } /** * 根据id删除数据 * @param index 索引 * @param id _id * @throws IOException */ public void DeleteEs(String index, String id) throws IOException { DeleteRequest deleteRequest = new DeleteRequest(index, id); DeleteResponse response = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT); log.info("删除数据至ElasticSearch成功:{}", response.getId()); } }
BinLogElasticSearch.javapackage com.example.canal.study.action; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.Message; import com.example.canal.study.common.CanalDataParser; import com.example.canal.study.common.ElasticUtils; import com.example.canal.study.pojo.Student; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; import java.io.IOException; import java.util.List; import java.util.Map; /** * @author haha */ @Slf4j @Component public class BinLogElasticSearch { @Autowired private CanalConnector canalSimpleConnector; @Autowired private ElasticUtils elasticUtils; //@Qualifier("canalHaConnector")使用名为canalHaConnector的bean @Autowired @Qualifier("canalHaConnector") private CanalConnector canalHaConnector; public void binLogToElasticSearch() throws IOException { openCanalConnector(canalHaConnector); // 轮询拉取数据 Integer batchSize = 5 * 1024; while (true) { Message message = canalHaConnector.getWithoutAck(batchSize); // Message message = canalSimpleConnector.getWithoutAck(batchSize); long id = message.getId(); int size = message.getEntries().size(); log.info("当前监控到binLog消息数量{}", size); if (id == -1 || size == 0) { try { // 等待2秒 Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } } else { //1. 解析message对象 List<CanalEntry.Entry> entries = message.getEntries(); List<CanalDataParser.TwoTuple<CanalEntry.EventType, Map>> rows = CanalDataParser.printEntry(entries); for (CanalDataParser.TwoTuple<CanalEntry.EventType, Map> tuple : rows) { if(tuple.eventType == CanalEntry.EventType.INSERT) { Student student = createStudent(tuple); // 2。将解析出的对象同步到elasticSearch中 elasticUtils.saveEs(student, "student_index"); // 3.消息确认已处理 // canalSimpleConnector.ack(id); canalHaConnector.ack(id); } if(tuple.eventType == CanalEntry.EventType.UPDATE){ Student student = createStudent(tuple); elasticUtils.updateEs(student, "student_index"); // 3.消息确认已处理 // canalSimpleConnector.ack(id); canalHaConnector.ack(id); } if(tuple.eventType == CanalEntry.EventType.DELETE){ elasticUtils.DeleteEs("student_index", tuple.columnMap.get("id").toString()); canalHaConnector.ack(id); } } } } } /** * 封装数据至Student * @param tuple * @return */ private Student createStudent(CanalDataParser.TwoTuple<CanalEntry.EventType, Map> tuple){ Student student = new Student(); student.setId(tuple.columnMap.get("id").toString()); student.setAge(Integer.parseInt(tuple.columnMap.get("age").toString())); student.setName(tuple.columnMap.get("name").toString()); student.setSex(tuple.columnMap.get("sex").toString()); student.setCity(tuple.columnMap.get("city").toString()); return student; } /** * 打开canal连接 * * @param canalConnector */ private void openCanalConnector(CanalConnector canalConnector) { //连接CanalServer canalConnector.connect(); // 订阅destination canalConnector.subscribe(); } /** * 关闭canal连接 * * @param canalConnector */ private void closeCanalConnector(CanalConnector canalConnector) { //关闭连接CanalServer canalConnector.disconnect(); // 注销订阅destination canalConnector.unsubscribe(); } }
CanalDemoApplication.java(Spring Boot启动类)package com.example.canal.study; import com.example.canal.study.action.BinLogElasticSearch; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; /** * @author haha */ @SpringBootApplication public class CanalDemoApplication implements ApplicationRunner { @Autowired private BinLogElasticSearch binLogElasticSearch; public static void main(String[] args) { SpringApplication.run(CanalDemoApplication.class, args); } // 程序启动则执行run方法 @Override public void run(ApplicationArguments args) throws Exception { binLogElasticSearch.binLogToElasticSearch(); } }
application.propertiesserver.port=8081 spring.application.name = canal-demo canal.server.ip = 192.168.124.5 canal.server.port = 11111 canal.destination = example zookeeper.server.ip = 192.168.124.5:2181 zookeeper.sasl.client = false elasticSearch.server.ip = 192.168.124.5 elasticSearch.server.port = 9200
Canal集群高可用的搭建通过上面的学习,我们知道了单机直连方式的Canala应用。在当今互联网时代,单实例模式逐渐被集群高可用模式取代,那么Canala的多实例集群方式如何搭建呢! 基于ZooKeeper获取Canal实例准备ZooKeeper的Docker镜像与容器: docker pull zookeeper docker run -d --name zookeeper --net mynetwork --ip 172.18.0.3 -p 2181:2181 zookeeper docker run -d --name canal-server2 --net mynetwork --ip 172.18.0.8 -p 11113:11113 canal/canal-server
canal.port=11113 canal.zkServers=172.18.0.3:2181 canal.instance.global.spring.xml = classpath:spring/default-instance.xml
canal.instance.mysql.slaveId = 1235 #之前的canal slaveId是1234,保证slaveId不重复即可 canal.instance.master.address = 172.18.0.6:3306
[zk: localhost:2181(CONNECTED) 15] get /otter/canal/destinations/example/running {"active":true,"address":"172.18.0.4:11111","cid":1}
客户端链接, 消费数据可以通过指定ZooKeeper地址和Canal的instance name,canal client会自动从ZooKeeper中的running节点获取当前服务的工作节点,然后与其建立链接: [zk: localhost:2181(CONNECTED) 0] get /otter/canal/destinations/example/running {"active":true,"address":"172.18.0.4:11111","cid":1}
CanalConnector connector = CanalConnectors.newClusterConnector("172.18.0.3:2181", "example", "", "");
[zk: localhost:2181(CONNECTED) 4] get /otter/canal/destinations/example/1001/running {"active":true,"address":"192.168.124.5:59887","clientId":1001}
[zk: localhost:2181(CONNECTED) 5] get /otter/canal/destinations/example/1001/cursor {"@type":"com.alibaba.otter.canal.protocol.position.LogPosition","identity":{"slaveId":-1,"sourceAddress":{"address":"mysql.mynetwork","port":3306}},"postion":{"included":false,"journalName":"binlog.000004","position":2169,"timestamp":1562672817000}}
docker exec -it canal-server bash cd canal-server/bin sh stop.sh
[zk: localhost:2181(CONNECTED) 19] get /otter/canal/destinations/example/running {"active":true,"address":"172.18.0.8:11111","cid":1}
异常与总结elasticsearch-head无法访问Elasticsearches与es-head是两个独立的进程,当es-head访问es服务时,会存在一个跨域问题。所以我们需要修改es的配置文件,增加一些配置项来解决这个问题,如下: [root@localhost /usr/local/elasticsearch-head-master]# cd ../elasticsearch-5.5.2/config/ [root@localhost /usr/local/elasticsearch-5.5.2/config]# vim elasticsearch.yml # 文件末尾加上如下配置 http.cors.enabled: true http.cors.allow-origin: "*"
elasticsearch-head查询报406 Not Acceptable
#6886行 contentType: "application/x-www-form-urlencoded 改成 contentType: "application/json;charset=UTF-8" #7574行 var inspectData = s.contentType === "application/x-www-form-urlencoded" && 改成 var inspectData = s.contentType === "application/json;charset=UTF-8" &&
使用elasticsearch-rest-high-level-client报org.elasticsearch.action.index.IndexRequest.ifSeqNo#pom中除了加入依赖 <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-high-level-client</artifactId> <version>7.1.1</version> </dependency> #还需加入 <dependency> <groupId>org.elasticsearch</groupId> <artifactId>elasticsearch</artifactId> <version>7.1.1</version> </dependency>
为什么ElasticSearch要在7.X版本不能使用type?
参考:为什么ElasticSearch要在7.X版本去掉type? 使用spring-data-elasticsearch.jar报org.elasticsearch.client.transport.NoNodeAvailableException由于本文使用的是elasticsearch7.x以上的版本,目前spring-data-elasticsearch底层采用es官方TransportClient,而es官方计划放弃TransportClient,工具以es官方推荐的RestHighLevelClient进行调用请求。 可参考RestHighLevelClient API。 设置Docker容器开启启动如果创建时未指定 --restart=always ,可通过update 命令 docker update --restart=always [containerID]
Docker for Mac network host模式不生效
Host模式是为了性能,但是这却对Docker的隔离性造成了破坏,导致安全性降低。 在性能场景下,可以用--netwokr host开启Host模式,但需要注意的是,如果你用Windows或Mac本地启动容器的话,会遇到Host模式失效的问题。原因是Host模式只支持Linux宿主机。 客户端连接ZooKeeper报authenticate using SASL(unknow error)
如果更换canal.client.jar中依赖的zookeeper.jar的版本把Canal的官方源码下载到本机git clone https://github.com/alibaba/canal.git,然后修改client模块下pom.xml文件中关于ZooKeeper的内容,然后重新mvn install:
关于选型的取舍 |
|