经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 程序设计 » Docker » 查看文章
基于Docker结合Canal实现MySQL实时增量数据传输功能
来源：jb51　　时间：2019/7/23 13:13:13　　对本文有异议

Canal的介绍

Canal的历史由来

在早期的时候，阿里巴巴公司因为杭州和美国两个地方的机房都部署了数据库实例，但因为跨机房同步数据的业务需求 ，便孕育而生出了Canal，主要是基于trigger（触发器）的方式获取增量变更。从2010年开始，阿里巴巴公司开始逐步尝试数据库日志解析，获取增量变更的数据进行同步，由此衍生出了增量订阅和消费业务。

当前的Canal支持的数据源端MySQL版本包括：5.1.x 、5.5.x 、5.6.x、5.7.x、8.0.x。

Canal的应用场景

目前普遍基于日志增量订阅和消费的业务，主要包括：

  1. 基于数据库增量日志解析，提供增量数据订阅和消费
  2. 数据库镜像 数据库实时备份
  3. 索引构建和实时维护（拆分异构索引、倒排索引等）
  4. 业务Cache刷新
  5. 带业务逻辑的增量数据处理
  6. Canal的工作原理

在介绍Canal的原理之前，我们先来了解下MySQL主从复制的原理。

MySQL主从复制原理

  • MySQL Master将数据变更的操作写入二进制日志binary log中， 其中记录的内容叫做二进制日志事件binary log events，可以通过show binlog events命令进行查看
  • MySQL Slave会将Master的binary log中的binary log events拷贝到它的中继日志relay log
  • MySQL Slave重读并执行relay log中的事件，将数据变更映射到它自己的数据库表中

了解了MySQL的工作原理，我们可以大致猜想到Canal应该也是采用类似的逻辑去实现增量数据订阅的功能，那么接下来我们看看实际上Canal的工作原理是怎样的？

Canal工作原理

  • Canal模拟MySQL Slave的交互协议，伪装自己为MySQL Slave，向MySQL Master发送dump协议
  • MySQL Master收到dump请求，开始推送binary log给Slave（也就是Canal）
  • Canal解析binary log对象（数据为byte流）

基于这样的原理与方式，便可以完成数据库增量日志的获取解析，提供增量数据订阅和消费，实现MySQL实时增量数据传输的功能。

既然Canal是这样的一个框架，又是纯Java语言编写而成，那么我们接下来就开始学习怎么使用它并把它用到我们的实际工作中。

Canal的Docker环境准备

因为目前容器化技术的火热，本文通过使用Docker来快速搭建开发环境，而传统方式的环境搭建，在我们学会了Docker容器环境搭建后，也能自行依葫芦画瓢搭建成功。由于本篇主要讲解Canal，所以关于Docker的内容不会涉及太多，主要会介绍Docker的基本概念和命令使用。 如果你想和更多容器技术专家交流，可以加我微信liyingjiese，备注『加群』。群里每周都有全球各大公司的最佳实践以及行业最新动态 。

什么是Docker

相信绝大多数人都使用过虚拟机VMware，在使用VMware进行环境搭建的时候，只需提供了一个普通的系统镜像并成功安装，剩下的软件环境与应用配置还是如我们在本机操作一样在虚拟机里也操作一遍，而且VMware占用宿主机的资源较多，容易造成宿主机卡顿，而且系统镜像本身也占用过多空间。

为了便于大家快速理解Docker，便与VMware做对比来做介绍，Docker提供了一个开始，打包，运行APP的平台，把APP（应用）和底层infrastructure（基础设施）隔离开来。Docker中最主要的两个概念就是镜像（类似VMware的系统镜像）与容器（类似VMware里安装的系统）。

什么是Image（镜像）

  • 文件和meta data的集合（root filesystem）
  • 分层的，并且每一层都可以添加改变删除文件，成为一个新的image
  • 不同的image可以共享相同的layer
  • Image本身是read-only的

什么是Container（容器）

  • 通过Image创建（copy）
  • 在Image layer之上建立一个container layer（可读写）
  • 类比面向对象：类和实例
  • Image负责APP的存储和分发，Container负责运行APP

Docker的网络介绍

Docker的网络类型有三种：

  • Bridge：桥接网络。默认情况下启动的Docker容器，都是使用Bridge，Docker安装时创建的桥接网络，每次Docker容器重启时，会按照顺序获取对应的IP地址，这个就导致重启下，Docker的IP地址就变了。
  • None：无指定网络。使用 --network=none，Docker容器就不会分配局域网的IP。
  • Host：主机网络。使用--network=host，此时，Docker容器的网络会附属在主机上，两者是互通的。例如，在容器中运行一个Web服务，监听8080端口，则主机的8080端口就会自动映射到容器中。

创建自定义网络：（设置固定IP）

  1. docker network create --subnet=172.18.0.0/16 mynetwork

查看存在的网络类型docker network ls：

搭建Canal环境

附上Docker的下载安装地址==> Docker Download 。

下载Canal镜像docker pull canal/canal-server：

下载MySQL镜像docker pull mysql，下载过的则如下图：

查看已经下载好的镜像docker images：

接下来通过镜像生成MySQL容器与canal-server容器：

  1. ##生成mysql容器
  2. docker run -d --name mysql --net mynetwork --ip 172.18.0.6 -p 3306:3306 -e MYSQL_ROOT_PASSWORD=root mysql
  3. ##生成canal-server容器
  4. docker run -d --name canal-server --net mynetwork --ip 172.18.0.4 -p 11111:11111 canal/canal-server
  5. ## 命令介绍
  6. --net mynetwork #使用自定义网络
  7. --ip #指定分配ip

查看Docker中运行的容器docker ps：

MySQL的配置修改

以上只是初步准备好了基础的环境，但是怎么让Canal伪装成Salve并正确获取MySQL中的binary log呢？

对于自建MySQL，需要先开启Binlog写入功能，配置binlog-format为ROW模式，通过修改MySQL配置文件来开启bin_log，使用find / -name my.cnf查找my.cnf，修改文件内容如下：

  1. [mysqld]
  2. log-bin=mysql-bin # 开启binlog
  3. binlog-format=ROW # 选择ROW模式
  4. server_id=1 # 配置MySQL replaction需要定义，不要和Canal的slaveId重复

进入MySQL容器docker exec -it mysql bash。

创建链接MySQL的账号Canal并授予作为MySQL slave的权限，如果已有账户可直接GRANT：

  1. mysql -uroot -proot
  2. # 创建账号
  3. CREATE USER canal IDENTIFIED BY 'canal'; 
  4. # 授予权限
  5. GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
  6. -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
  7. # 刷新并应用
  8. FLUSH PRIVILEGES;

数据库重启后，简单测试 my.cnf 配置是否生效：

  1. show variables like 'log_bin';
  2. show variables like 'log_bin';
  3. show master status;

canal-server的配置修改

进入canal-server容器docker exec -it canal-server bash。

编辑canal-server的配置vi canal-server/conf/example/instance.properties：

更多配置请参考==>Canal配置说明 。

重启canal-server容器docker restart canal-server 进入容器查看启动日志：

  1. docker exec -it canal-server bash
  2. tail -100f canal-server/logs/example/example.log

至此，我们的环境工作准备完成！

拉取数据并同步保存到ElasticSearch

本文的ElasticSearch也是基于Docker环境搭建，所以读者可执行如下命令：

  1. # 下载对镜像
  2. docker pull elasticsearch:7.1.1
  3. docker pull mobz/elasticsearch-head:5-alpine
  4. # 创建容器并运行
  5. 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
  6. docker run -d --name elasticsearch-head --net mynetwork --ip 172.18.0.5 -p 9100:9100 mobz/elasticsearch-head:5-alpine

环境已经准备好了，现在就要开始我们的编码实战部分了，怎么通过应用程序去获取Canal解析后的binlog数据。首先我们基于Spring Boot搭建一个canal demo应用。结构如下图所示：

Student.java

  1. package com.example.canal.study.pojo;
  2. import lombok.Data;
  3. import java.io.Serializable;
  4. // @Data 用户生产getter、setter方法
  5. @Data
  6. public class Student implements Serializable {
  7. private String id;
  8. private String name;
  9. private int age;
  10. private String sex;
  11. private String city;
  12. }

CanalConfig.java

  1. package com.example.canal.study.common;
  2. import com.alibaba.otter.canal.client.CanalConnector;
  3. import com.alibaba.otter.canal.client.CanalConnectors;
  4. import org.apache.http.HttpHost;
  5. import org.elasticsearch.client.RestClient;
  6. import org.elasticsearch.client.RestHighLevelClient;
  7. import org.springframework.beans.factory.annotation.Value;
  8. import org.springframework.context.annotation.Bean;
  9. import org.springframework.context.annotation.Configuration;
  10. import java.net.InetSocketAddress;
  11. /**
  12. * @author haha
  13. */
  14. @Configuration
  15. public class CanalConfig {
  16. // @Value 获取 application.properties配置中端内容
  17. @Value("${canal.server.ip}")
  18. private String canalIp;
  19. @Value("${canal.server.port}")
  20. private Integer canalPort;
  21. @Value("${canal.destination}")
  22. private String destination;
  23. @Value("${elasticSearch.server.ip}")
  24. private String elasticSearchIp;
  25. @Value("${elasticSearch.server.port}")
  26. private Integer elasticSearchPort;
  27. @Value("${zookeeper.server.ip}")
  28. private String zkServerIp;
  29. // 获取简单canal-server连接
  30. @Bean
  31. public CanalConnector canalSimpleConnector() {
  32.  CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalIp, canalPort), destination, "", "");
  33.  return canalConnector;
  34. }
  35. // 通过连接zookeeper获取canal-server连接
  36. @Bean
  37. public CanalConnector canalHaConnector() {
  38.  CanalConnector canalConnector = CanalConnectors.newClusterConnector(zkServerIp, destination, "", "");
  39.  return canalConnector;
  40. }
  41. // elasticsearch 7.x客户端
  42. @Bean
  43. public RestHighLevelClient restHighLevelClient() {
  44.  RestHighLevelClient client = new RestHighLevelClient(
  45.    RestClient.builder(new HttpHost(elasticSearchIp, elasticSearchPort))
  46.  );
  47.  return client;
  48. }
  49. }

CanalDataParser.java

由于这个类的代码较多，文中则摘出其中比较重要的部分，其它部分代码可从GitHub上获取：

  1. public static class TwoTuple<A, B> {
  2.  public final A eventType;
  3.  public final B columnMap;
  4.  public TwoTuple(A a, B b) {
  5.   eventType = a;
  6.   columnMap = b;
  7.  }
  8. }
  9. public static List<TwoTuple<EventType, Map>> printEntry(List<Entry> entrys) {
  10.  List<TwoTuple<EventType, Map>> rows = new ArrayList<>();
  11.  for (Entry entry : entrys) {
  12.   // binlog event的事件事件
  13.   long executeTime = entry.getHeader().getExecuteTime();
  14.   // 当前应用获取到该binlog锁延迟的时间
  15.   long delayTime = System.currentTimeMillis() - executeTime;
  16.   Date date = new Date(entry.getHeader().getExecuteTime());
  17.   SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  18.   // 当前的entry（binary log event）的条目类型属于事务
  19.   if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
  20.    if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN) {
  21.     TransactionBegin begin = null;
  22.     try {
  23.      begin = TransactionBegin.parseFrom(entry.getStoreValue());
  24.     } catch (InvalidProtocolBufferException e) {
  25.      throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
  26.     }
  27.     // 打印事务头信息，执行的线程id，事务耗时
  28.     logger.info(transaction_format,
  29.       new Object[]{entry.getHeader().getLogfileName(),
  30.         String.valueOf(entry.getHeader().getLogfileOffset()),
  31.         String.valueOf(entry.getHeader().getExecuteTime()),
  32.         simpleDateFormat.format(date),
  33.         entry.getHeader().getGtid(),
  34.         String.valueOf(delayTime)});
  35.     logger.info(" BEGIN ----> Thread id: {}", begin.getThreadId());
  36.     printXAInfo(begin.getPropsList());
  37.    } else if (entry.getEntryType() == EntryType.TRANSACTIONEND) {
  38.     TransactionEnd end = null;
  39.     try {
  40.      end = TransactionEnd.parseFrom(entry.getStoreValue());
  41.     } catch (InvalidProtocolBufferException e) {
  42.      throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
  43.     }
  44.     // 打印事务提交信息，事务id
  45.     logger.info("----------------\n");
  46.     logger.info(" END ----> transaction id: {}", end.getTransactionId());
  47.     printXAInfo(end.getPropsList());
  48.     logger.info(transaction_format,
  49.       new Object[]{entry.getHeader().getLogfileName(),
  50.         String.valueOf(entry.getHeader().getLogfileOffset()),
  51.         String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
  52.         entry.getHeader().getGtid(), String.valueOf(delayTime)});
  53.    }
  54.    continue;
  55.   }
  56.   // 当前entry（binary log event）的条目类型属于原始数据
  57.   if (entry.getEntryType() == EntryType.ROWDATA) {
  58.    RowChange rowChage = null;
  59.    try {
  60.     // 获取储存的内容
  61.     rowChage = RowChange.parseFrom(entry.getStoreValue());
  62.    } catch (Exception e) {
  63.     throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
  64.    }
  65.    // 获取当前内容的事件类型
  66.    EventType eventType = rowChage.getEventType();
  67.    logger.info(row_format,
  68.      new Object[]{entry.getHeader().getLogfileName(),
  69.        String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(),
  70.        entry.getHeader().getTableName(), eventType,
  71.        String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
  72.        entry.getHeader().getGtid(), String.valueOf(delayTime)});
  73.    // 事件类型是query或数据定义语言DDL直接打印sql语句，跳出继续下一次循环
  74.    if (eventType == EventType.QUERY || rowChage.getIsDdl()) {
  75.     logger.info(" sql ----> " + rowChage.getSql() + SEP);
  76.     continue;
  77.    }
  78.    printXAInfo(rowChage.getPropsList());
  79.    // 循环当前内容条目的具体数据
  80.    for (RowData rowData : rowChage.getRowDatasList()) {
  81.     List<CanalEntry.Column> columns;
  82.     // 事件类型是delete返回删除前的列内容，否则返回改变后列的内容
  83.     if (eventType == CanalEntry.EventType.DELETE) {
  84.      columns = rowData.getBeforeColumnsList();
  85.     } else {
  86.      columns = rowData.getAfterColumnsList();
  87.     }
  88.     HashMap<String, Object> map = new HashMap<>(16);
  89.     // 循环把列的name与value放入map中
  90.     for (Column column: columns){
  91.      map.put(column.getName(), column.getValue());
  92.     }
  93.     rows.add(new TwoTuple<>(eventType, map));
  94.    }
  95.   }
  96.  }
  97.  return rows;
  98. }

ElasticUtils.java

  1. package com.example.canal.study.common;
  2. import com.alibaba.fastjson.JSON;
  3. import com.example.canal.study.pojo.Student;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.elasticsearch.client.RestHighLevelClient;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.stereotype.Component;
  8. import org.elasticsearch.action.DocWriteRequest;
  9. import org.elasticsearch.action.delete.DeleteRequest;
  10. import org.elasticsearch.action.delete.DeleteResponse;
  11. import org.elasticsearch.action.get.GetRequest;
  12. import org.elasticsearch.action.get.GetResponse;
  13. import org.elasticsearch.action.index.IndexRequest;
  14. import org.elasticsearch.action.index.IndexResponse;
  15. import org.elasticsearch.action.update.UpdateRequest;
  16. import org.elasticsearch.action.update.UpdateResponse;
  17. import org.elasticsearch.client.RequestOptions;
  18. import org.elasticsearch.common.xcontent.XContentType;
  19. import java.io.IOException;
  20. import java.util.Map;
  21. /**
  22. * @author haha
  23. */
  24. @Slf4j
  25. @Component
  26. public class ElasticUtils {
  27. @Autowired
  28. private RestHighLevelClient restHighLevelClient;
  29. /**
  30.  * 新增
  31.  * @param student 
  32.  * @param index 索引
  33.  */
  34. public void saveEs(Student student, String index) {
  35.  IndexRequest indexRequest = new IndexRequest(index)
  36.    .id(student.getId())
  37.    .source(JSON.toJSONString(student), XContentType.JSON)
  38.    .opType(DocWriteRequest.OpType.CREATE);
  39.  try {
  40.   IndexResponse response = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
  41.   log.info("保存数据至ElasticSearch成功：{}", response.getId());
  42.  } catch (IOException e) {
  43.   log.error("保存数据至elasticSearch失败: {}", e);
  44.  }
  45. }
  46. /**
  47.  * 查看
  48.  * @param index 索引
  49.  * @param id _id
  50.  * @throws IOException
  51.  */
  52. public void getEs(String index, String id) throws IOException {
  53.  GetRequest getRequest = new GetRequest(index, id);
  54.  GetResponse response = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
  55.  Map<String, Object> fields = response.getSource();
  56.  for (Map.Entry<String, Object> entry : fields.entrySet()) {
  57.   System.out.println(entry.getKey() + ":" + entry.getValue());
  58.  }
  59. }
  60. /**
  61.  * 更新
  62.  * @param student
  63.  * @param index 索引
  64.  * @throws IOException
  65.  */
  66. public void updateEs(Student student, String index) throws IOException {
  67.  UpdateRequest updateRequest = new UpdateRequest(index, student.getId());
  68.  updateRequest.upsert(JSON.toJSONString(student), XContentType.JSON);
  69.  UpdateResponse response = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
  70.  log.info("更新数据至ElasticSearch成功：{}", response.getId());
  71. }
  72. /**
  73.  * 根据id删除数据
  74.  * @param index 索引
  75.  * @param id _id
  76.  * @throws IOException
  77.  */
  78. public void DeleteEs(String index, String id) throws IOException {
  79.  DeleteRequest deleteRequest = new DeleteRequest(index, id);
  80.  DeleteResponse response = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
  81.  log.info("删除数据至ElasticSearch成功：{}", response.getId());
  82. }
  83. }

BinLogElasticSearch.java

  1. package com.example.canal.study.action;
  2. import com.alibaba.otter.canal.client.CanalConnector;
  3. import com.alibaba.otter.canal.protocol.CanalEntry;
  4. import com.alibaba.otter.canal.protocol.Message;
  5. import com.example.canal.study.common.CanalDataParser;
  6. import com.example.canal.study.common.ElasticUtils;
  7. import com.example.canal.study.pojo.Student;
  8. import lombok.extern.slf4j.Slf4j;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.beans.factory.annotation.Qualifier;
  11. import org.springframework.stereotype.Component;
  12. import java.io.IOException;
  13. import java.util.List;
  14. import java.util.Map;
  15. /**
  16. * @author haha
  17. */
  18. @Slf4j
  19. @Component
  20. public class BinLogElasticSearch {
  21. @Autowired
  22. private CanalConnector canalSimpleConnector;
  23. @Autowired
  24. private ElasticUtils elasticUtils;
  25. //@Qualifier("canalHaConnector")使用名为canalHaConnector的bean
  26. @Autowired
  27. @Qualifier("canalHaConnector")
  28. private CanalConnector canalHaConnector;
  29. public void binLogToElasticSearch() throws IOException {
  30.  openCanalConnector(canalHaConnector);
  31.  // 轮询拉取数据
  32.  Integer batchSize = 5 * 1024;
  33.  while (true) {
  34.   Message message = canalHaConnector.getWithoutAck(batchSize);
  35. //   Message message = canalSimpleConnector.getWithoutAck(batchSize);
  36.   long id = message.getId();
  37.   int size = message.getEntries().size();
  38.   log.info("当前监控到binLog消息数量{}", size);
  39.   if (id == -1 || size == 0) {
  40.    try {
  41.     // 等待2秒
  42.     Thread.sleep(2000);
  43.    } catch (InterruptedException e) {
  44.     e.printStackTrace();
  45.    }
  46.   } else {
  47.    //1. 解析message对象
  48.    List<CanalEntry.Entry> entries = message.getEntries();
  49.    List<CanalDataParser.TwoTuple<CanalEntry.EventType, Map>> rows = CanalDataParser.printEntry(entries);
  50.    for (CanalDataParser.TwoTuple<CanalEntry.EventType, Map> tuple : rows) {
  51.     if(tuple.eventType == CanalEntry.EventType.INSERT) {
  52.      Student student = createStudent(tuple);
  53.      // 2。将解析出的对象同步到elasticSearch中
  54.      elasticUtils.saveEs(student, "student_index");
  55.      // 3.消息确认已处理
  56. //     canalSimpleConnector.ack(id);
  57.      canalHaConnector.ack(id);
  58.     }
  59.     if(tuple.eventType == CanalEntry.EventType.UPDATE){
  60.      Student student = createStudent(tuple);
  61.      elasticUtils.updateEs(student, "student_index");
  62.      // 3.消息确认已处理
  63. //     canalSimpleConnector.ack(id);
  64.      canalHaConnector.ack(id);
  65.     }
  66.     if(tuple.eventType == CanalEntry.EventType.DELETE){
  67.      elasticUtils.DeleteEs("student_index", tuple.columnMap.get("id").toString());
  68.      canalHaConnector.ack(id);
  69.     }
  70.    }
  71.   }
  72.  }
  73. }
  74. /**
  75.  * 封装数据至Student
  76.  * @param tuple
  77.  * @return
  78.  */
  79. private Student createStudent(CanalDataParser.TwoTuple<CanalEntry.EventType, Map> tuple){
  80.  Student student = new Student();
  81.  student.setId(tuple.columnMap.get("id").toString());
  82.  student.setAge(Integer.parseInt(tuple.columnMap.get("age").toString()));
  83.  student.setName(tuple.columnMap.get("name").toString());
  84.  student.setSex(tuple.columnMap.get("sex").toString());
  85.  student.setCity(tuple.columnMap.get("city").toString());
  86.  return student;
  87. }
  88. /**
  89.  * 打开canal连接
  90.  *
  91.  * @param canalConnector
  92.  */
  93. private void openCanalConnector(CanalConnector canalConnector) {
  94.  //连接CanalServer
  95.  canalConnector.connect();
  96.  // 订阅destination
  97.  canalConnector.subscribe();
  98. }
  99. /**
  100.  * 关闭canal连接
  101.  *
  102.  * @param canalConnector
  103.  */
  104. private void closeCanalConnector(CanalConnector canalConnector) {
  105.  //关闭连接CanalServer
  106.  canalConnector.disconnect();
  107.  // 注销订阅destination
  108.  canalConnector.unsubscribe();
  109. }
  110. }

CanalDemoApplication.java（Spring Boot启动类）

  1. package com.example.canal.study;
  2. import com.example.canal.study.action.BinLogElasticSearch;
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.boot.ApplicationArguments;
  5. import org.springframework.boot.ApplicationRunner;
  6. import org.springframework.boot.SpringApplication;
  7. import org.springframework.boot.autoconfigure.SpringBootApplication;
  8. /**
  9. * @author haha
  10. */
  11. @SpringBootApplication
  12. public class CanalDemoApplication implements ApplicationRunner {
  13. @Autowired
  14. private BinLogElasticSearch binLogElasticSearch;
  15. public static void main(String[] args) {
  16.  SpringApplication.run(CanalDemoApplication.class, args);
  17. }
  18. // 程序启动则执行run方法
  19. @Override
  20. public void run(ApplicationArguments args) throws Exception {
  21.  binLogElasticSearch.binLogToElasticSearch();
  22. }
  23. }

application.properties

  1. server.port=8081
  2. spring.application.name = canal-demo
  3. canal.server.ip = 192.168.124.5
  4. canal.server.port = 11111
  5. canal.destination = example
  6. zookeeper.server.ip = 192.168.124.5:2181
  7. zookeeper.sasl.client = false
  8. elasticSearch.server.ip = 192.168.124.5
  9. elasticSearch.server.port = 9200

Canal集群高可用的搭建

通过上面的学习，我们知道了单机直连方式的Canala应用。在当今互联网时代，单实例模式逐渐被集群高可用模式取代，那么Canala的多实例集群方式如何搭建呢！

基于ZooKeeper获取Canal实例

准备ZooKeeper的Docker镜像与容器：

  1. docker pull zookeeper
  2. docker run -d --name zookeeper --net mynetwork --ip 172.18.0.3 -p 2181:2181 zookeeper
  3. docker run -d --name canal-server2 --net mynetwork --ip 172.18.0.8 -p 11113:11113 canal/canal-server

1、机器准备：

  • 运行Canal的容器IP： 172.18.0.4 , 172.18.0.8
  • ZooKeeper容器IP：172.18.0.3:2181
  • MySQL容器IP：172.18.0.6:3306

2、按照部署和配置，在单台机器上各自完成配置，演示时instance name为example。

3、修改canal.properties，加上ZooKeeper配置并修改Canal端口：

  1. canal.port=11113
  2. canal.zkServers=172.18.0.3:2181
  3. canal.instance.global.spring.xml = classpath:spring/default-instance.xml

4、创建example目录，并修改instance.properties：

  1. canal.instance.mysql.slaveId = 1235 
  2. #之前的canal slaveId是1234，保证slaveId不重复即可
  3. canal.instance.master.address = 172.18.0.6:3306

注意： 两台机器上的instance目录的名字需要保证完全一致，HA模式是依赖于instance name进行管理，同时必须都选择default-instance.xml配置。

启动两个不同容器的Canal，启动后，可以通过tail -100f logs/example/example.log查看启动日志，只会看到一台机器上出现了启动成功的日志。

比如我这里启动成功的是 172.18.0.4：

查看一下ZooKeeper中的节点信息，也可以知道当前工作的节点为172.18.0.4:11111：

  1. [zk: localhost:2181(CONNECTED) 15] get /otter/canal/destinations/example/running 
  2. {"active":true,"address":"172.18.0.4:11111","cid":1}

客户端链接, 消费数据

可以通过指定ZooKeeper地址和Canal的instance name，canal client会自动从ZooKeeper中的running节点获取当前服务的工作节点，然后与其建立链接：

  1. [zk: localhost:2181(CONNECTED) 0] get /otter/canal/destinations/example/running
  2. {"active":true,"address":"172.18.0.4:11111","cid":1}

对应的客户端编码可以使用如下形式，上文中的CanalConfig.java中的canalHaConnector就是一个HA连接：

  1. CanalConnector connector = CanalConnectors.newClusterConnector("172.18.0.3:2181", "example", "", "");

链接成功后，canal server会记录当前正在工作的canal client信息，比如客户端IP，链接的端口信息等（聪明的你，应该也可以发现，canal client也可以支持HA功能）：

  1. [zk: localhost:2181(CONNECTED) 4] get /otter/canal/destinations/example/1001/running
  2. {"active":true,"address":"192.168.124.5:59887","clientId":1001}

数据消费成功后，canal server会在ZooKeeper中记录下当前最后一次消费成功的binlog位点（下次你重启client时，会从这最后一个位点继续进行消费）：

  1. [zk: localhost:2181(CONNECTED) 5] get /otter/canal/destinations/example/1001/cursor
  2.  
  3. {"@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}}

停止正在工作的172.18.0.4的canal server：

  1. docker exec -it canal-server bash
  2. cd canal-server/bin
  3. sh stop.sh

这时172.18.0.8会立马启动example instance，提供新的数据服务：

  1. [zk: localhost:2181(CONNECTED) 19] get /otter/canal/destinations/example/running
  2. {"active":true,"address":"172.18.0.8:11111","cid":1}

与此同时，客户端也会随着canal server的切换，通过获取ZooKeeper中的最新地址，与新的canal server建立链接，继续消费数据，整个过程自动完成。

异常与总结

elasticsearch-head无法访问Elasticsearch

es与es-head是两个独立的进程，当es-head访问es服务时，会存在一个跨域问题。所以我们需要修改es的配置文件，增加一些配置项来解决这个问题，如下：

  1. [root@localhost /usr/local/elasticsearch-head-master]# cd ../elasticsearch-5.5.2/config/
  2. [root@localhost /usr/local/elasticsearch-5.5.2/config]# vim elasticsearch.yml 
  3. # 文件末尾加上如下配置
  4. http.cors.enabled: true
  5. http.cors.allow-origin: "*"

修改完配置文件后需重启es服务。

elasticsearch-head查询报406 Not Acceptable

解决方法：

1、进入head安装目录；

2、cd _site/

3、编辑vendor.js 共有两处

  1. #6886行 contentType: "application/x-www-form-urlencoded
  2. 改成 contentType: "application/json;charset=UTF-8"
  3.  #7574行 var inspectData = s.contentType === "application/x-www-form-urlencoded" &&
  4. 改成 var inspectData = s.contentType === "application/json;charset=UTF-8" &&

使用elasticsearch-rest-high-level-client报org.elasticsearch.action.index.IndexRequest.ifSeqNo

  1. #pom中除了加入依赖
  2. <dependency>
  3. <groupId>org.elasticsearch.client</groupId>
  4. <artifactId>elasticsearch-rest-high-level-client</artifactId>
  5. <version>7.1.1</version>
  6. </dependency>
  7. #还需加入
  8. <dependency>
  9. <groupId>org.elasticsearch</groupId>
  10. <artifactId>elasticsearch</artifactId>
  11. <version>7.1.1</version>
  12. </dependency>

相关参考： git hub issues 。

为什么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容器开启启动

  1. 如果创建时未指定 --restart=always ,可通过update 命令
  2. docker update --restart=always [containerID]

Docker for Mac network host模式不生效

Host模式是为了性能，但是这却对Docker的隔离性造成了破坏，导致安全性降低。 在性能场景下，可以用--netwokr host开启Host模式，但需要注意的是，如果你用Windows或Mac本地启动容器的话，会遇到Host模式失效的问题。原因是Host模式只支持Linux宿主机。

参见官方文档：  https://docs.docker.com/network/host/  。

客户端连接ZooKeeper报authenticate using SASL（unknow error）

  • zookeeper.jar与Dokcer中的ZooKeeper版本不一致
  • zookeeper.jar使用了3.4.6之前的版本

出现这个错的意思是ZooKeeper作为外部应用需要向系统申请资源，申请资源的时候需要通过认证，而sasl是一种认证方式，我们想办法来绕过sasl认证。避免等待，来提高效率。

在项目代码中加入System.setProperty("zookeeper.sasl.client", "false");，如果是Spring Boot项目可以在application.properties中加入zookeeper.sasl.client=false。

参考： Increased CPU usage by unnecessary SASL checks 。

如果更换canal.client.jar中依赖的zookeeper.jar的版本

把Canal的官方源码下载到本机git clone  https://github.com/alibaba/canal.git  ，然后修改client模块下pom.xml文件中关于ZooKeeper的内容，然后重新mvn install：

把自己项目依赖的包替换为刚刚mvn install生产的包：

关于选型的取舍

总结

以上所述是小编给大家介绍的基于Docker结合Canal实现MySQL实时增量数据传输功能,希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对w3xue网站的支持！
如果你觉得本文对你有帮助，欢迎转载，烦请注明出处，谢谢！

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Docker热门文章

  • docker部署certbot与nginx来获取ssl证书添加https及自动更新
  • 解决docker的tls(ssl)证书过期问题
  • 人工智能,丹青圣手,全平台(原生/Docker)构建Stable-Diffusion-Webui的AI绘画库教程(Python3.10/Pytorch1.13.0)
  • Docker原理
  • docker&nbsp;for&nbsp;windonws之Windows&nbsp;10&nbsp;家庭中文版安装clickhouse&nbsp;22.3版本及配置过程
  • Docker-compose command 有多个命令例子
  • docker修改容器配置文件的3种方法总结
  • 解决docker pull镜像报错的问题
  • Docker在Centos 7上的部署
  • Docker使用Prune命令清理none镜像

    • Docker推荐文章

  • 如何在Linux云服务器上通过Docker Compose部署安装Halo，搭建个人博客网站？
  • 使用Cloudflare Worker加速docker镜像
  • 如何在 Windows 使用 Podman Desktop 取代 Docker Desktop
  • Linux 提权-Docker 容器
  • 配置pod拉取harbor容器镜像仓库私有镜像：secret保存账号密码
  • Win11系统下的MindSpore环境搭建
  • 白话理解和使用DOCKER VOLUME
  • Docker 必知必会4----容器之间的通信
  • Docker 必知必会3----使用自己制作的镜像
  • 对于Docker和Podman的一点使用经验

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号