还在为数据同步而苦恼吗?手把手教你实现canal数据同步(三)

Canal实战之客户端适配器

canal 1.1.1版本之后, 增加客户端数据落地的适配及启动功能, 目前支持功能:

• 客户端启动器
• 同步管理REST接口
• 日志适配器, 作为DEMO
• 关系型数据库的数据同步(表对表同步), ETL功能
• HBase的数据同步(表对表同步), ETL功能
• ElasticSearch多表数据同步,ETL功能（新）
• 后续支持redis、mongodb

适配器整体结构

client-adapter分为适配器和启动器两部分, 适配器为多个fat jar, 每个适配器会将自己所需的依赖打成一个包, 以SPI的方式让启动器动态加载, 目前所有支持的适配器都放置在plugin目录下。

启动器为 SpringBoot 项目, 运行目录结构为:

详细结构如下：

- bin
    restart.sh
    startup.bat
    startup.sh
    stop.sh
- conf
bootstrap.yml
    application.yml
    - es
    biz_order.yml
    customer.yml
    mytest_user.yml
    - hbase
        mytest_person2.yml
    - rdb
    mytest_user.yml
- lib
...
- logs
- plugin
client-adapter.elasticsearch-1.1.4-jar-with-dependencies.jar
    client-adapter.hbase-1.1.4-jar-with-dependencies.jar
    client-adapter.logger-1.1.4-jar-with-dependencies.jar
    client-adapter.rdb-1.1.4-jar-with-dependencies.jar

适配器配置介绍

总配置文件 application.yml：

canal.conf:
  canalServerHost: 127.0.0.1:11111     # 对应单机模式下的canal server的ip:port
  zookeeperHosts: slave1:2181          # 对应集群模式下的zk地址, 如果配置了canalServerHost, 则以canalServerHost为准
  mqServers: slave1:6667 #or rocketmq  # kafka或rocketMQ地址, 与canalServerHost不能并存
  flatMessage: true                    # 扁平message开关, 是否以json字符串形式投递数据, 仅在kafka/rocketMQ模式下有效
  batchSize: 50                        # 每次获取数据的批大小, 单位为K
  syncBatchSize: 1000                  # 每次同步的批数量
  retries: 0                           # 重试次数, -1为无限重试
  timeout:                             # 同步超时时间, 单位毫秒
  mode: tcp # kafka rocketMQ           # canal client的模式: tcp直连 kafka rocketMQ
  srcDataSources:                      # 源数据库
    defaultDS:                         # 自定义名称
      url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true   # jdbc url 
      username: root                                            # jdbc 账号
      password: 121212                                          # jdbc 密码
  canalAdapters:                       # 适配器列表
  - instance: example                  # canal 实例名或者 MQ topic 名
    groups:                            # 分组列表
    - groupId: g1                      # 分组id, 如果是MQ模式将用到该值
      outerAdapters:                   # 分组内适配器列表
      - name: logger                   # 日志打印适配器
......

说明:

1. 一份数据可以被多个group同时消费, 多个group之间会是一个并行执行, 一个group内部是一个串行执行多个outerAdapters, 比如例子中logger和hbase

2. 目前client adapter数据订阅的方式支持两种，直连canal server 或者 订阅kafka/RocketMQ的消息

适配器启动

前提：启动canal server (单机模式)

上传canal.adapter-1.1.4.tar.gz到/opt目录下

1. 在/opt目录下创建canal-adapter目录：mkdir canal-adapter

2. 解压压缩包：tar -zxvf canal.adapter-1.1.4.tar.gz -C canal-adapter

3. 修改conf/application.yml，如下

4. 启动：bin/startup.sh

server:
  port: 8081
logging:
  level:
    com.alibaba.otter.canal.client.adapter.hbase: DEBUG
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8
    default-property-inclusion: non_null


canal.conf:
  canalServerHost: 127.0.0.1:11111
  batchSize: 500                            
  syncBatchSize: 1000                       
  retries: 0                               
  timeout:                                 
  mode: tcp 
  canalAdapters:                            
  - instance: example                       
    groups:                                 
    - groupId: g1                           
      outerAdapters:                        
      - name: logger

logger适配器：

最简单的处理, 将收到的变更事件通过日志打印的方式进行输出, 如配置所示, 只需要定义name: logger即可。

adapter管理REST接口

查询所有订阅同步的canal instance或MQ topic：

curl http://127.0.0.1:8081/destinations

数据同步开关：

curl http://127.0.0.1:8081/syncSwitch/example/off -X PUT

针对 example 这个canal instance/MQ topic 进行开关操作. off代表关闭, instance/topic下的同步将阻塞或者断开连接不再接收数据, on代表开启

注: 如果在配置文件中配置了 zookeeperHosts 项, 则会使用分布式锁来控制HA中的数据同步开关, 如果是单机模式则使用本地锁来控制开关

数据同步开关状态：

curl http://127.0.0.1:8081/syncSwitch/example

查看指定 canal instance/MQ topic 的数据同步开关状态