一：canal同步原理

其实canal同步的原理就是把自己伪装为一个Slave,去解析获取到的二进制日志文件，通过适配器得到不同终端所需要的数据

二：mysql设置

• mysql日志文件配置（ROW： 记录每一行数据的变化（最详细的记录方式），记录的日志内容包括行数据的变更。这种格式是进行主从同步时的推荐格式，尤其适用于需要精确同步数据变更的情况。）

  SET GLOBAL binlog_format = 'ROW'; 
  

• 创建canal用户，用于进行数据同步

  CREATE USER canal IDENTIFIED BY 'canal';
  GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
  FLUSH PRIVILEGES;
  

三：JDK安装

注：需要注意与es的版本是否兼容，否则后面可能会报各式和样的错误（我这里es是docker安装所以java版本影响不是很大）

地址：https://www.oracle.com/java/technologies/downloads/#jdk23-windows

四：canal配置

• canal下载（选择自己需要的版本）

  https://github.com/alibaba/canal/releases/tag/canal-1.1.8-alpha-2

• canal.deployer配置

  

  ......
  canal.instance.master.address=127.0.0.2:3306 #mysql的连接地址
  ......
  # 上面提到的用于同步数据的用户名与密码
  canal.instance.dbUsername=canal
  canal.instance.dbPassword=canal
  ......
  

• 启动canal服务

  	#这是我canal的安装目录，具体替换为你canal.deployer的存放目录即可
  	cd /soft/canalDeployer/bin  
  	./startup.sh  #启动文件
  

  可以在/soft/canalDeployer/logs/example/example.log中看到successful 说明服务启动成功

  
  注：如果遇到日志中报有druid相关的错误，直接下载https://repo1.maven.org/maven2/com/alibaba/druid/1.2.22/druid-1.2.22.jar
  丢到/soft/canalDeployer/lib下完事

• canal.adapte 配置
  修改application.yml

  cd /soft/canalAdapter/conf
  

  修改后的application.yml

  server:
    port: 8081
  spring:
    jackson:
      date-format: yyyy-MM-dd HH:mm:ss
      time-zone: GMT+8
      default-property-inclusion: non_null
  
  canal.conf:
    mode: tcp #tcp kafka rocketMQ rabbitMQ
    flatMessage: true
    zookeeperHosts:
    syncBatchSize: 1000
    retries: -1
    timeout:
    accessKey:
    secretKey:
    consumerProperties:
      # canal tcp consumer
      canal.tcp.server.host: 127.0.0.1:11111 # 服务地址与端口，需要与/soft/canalDeployer/conf/canal.properties  文件中的canal.ip与canal.port保持对应
      canal.tcp.zookeeper.hosts:
      canal.tcp.batch.size: 500
      canal.tcp.username:
      canal.tcp.password:
    srcDataSources:
      defaultDS:
        url: jdbc:mysql://127.0.0.1:3306/dev?useUnicode=true # 数据源配置
        username: canal
        password: canal
    canalAdapters:
      - instance: example # canal instance Name or mq topic name
        groups:
          - groupId: g1
            outerAdapters:
              - name: logger
              - name: es8 # 注意这个name，如果用es7 就写es7，与/soft/canalAdapter/conf/ 下的es进行对应
                hosts: http://127.0.0.1:9200 # es连接信息
                properties:
                  suppport.old: true
                  mode: rest # or rest
                  security.auth: 用户名:密码#  only used for rest mode
                  cluster.name: elasticsearch
  
  

• 添加同步的配置文件

  cd /soft/canalAdapter/conf/es8
  vim test.yml
  

  dataSourceKey: defaultDS
  destination: example
  groupId: g1
  esMapping:
    _index: test
    _id: _id
    upsert: true
    pk: id
    sql: "select su.id as _id,su.user_name as user_name from sx_user su"
    #  objFields:
    #    _labels: array:;
  #   etlCondition: "where a.c_time>={}"
    commitBatch: 3000
  
  

  注意:表名与字段名的别名问题，如果写原生，可能会报"com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem$ColumnItem.getOwner()" is null 这样的错误，查看了好多资料才得到解决

• 添加ES索引

  PUT /test
  {
      "mappings" : {
        "properties" : {
          "id" : {
            "type" : "integer"
          },
          "user_name":{
            "type":"text"
          }
        }
      }
   }
  

• canal.adapte 启动

  cd /soft/canalAdapter/bin
  ./startup.sh
  

  不出意外的话会得到如下结果，证明你成功了
  

• 测试全量同步

  curl http://127.0.0.1:8081/etl/es8/test.yml -X POST
  

  得到结果

  {"succeeded":true,"resultMessage":"导入ES 数据：139 条"}
  

• 测试增量同步，在数据库中修改一条信息，查看日志

  
  查看es中数据

  GET test/_search
  {
    "query": {
      "term": {
        "_id": 1
      }
    }
  }
  

  
  祝贺大功告成