canal应用四：canal-adapter的使用

文章目录

前言
适配器结构
适配器配置说明
- 总配置文件 application.yml
- 使用远程配置（MySQL）
适配器启动

前言

前面的文章，我们都是通过编写Java程序来实现Client消费。Canal 提供了几种适配器，允许我们通过一定的配置实现消费数据并同步到其他存储介质。目前支持功能：

日志适配器
关系型数据库的数据同步(表对表同步), ETL功能
Hbase的数据同步(表对表同步), ETL功能
Elasticsearch的数据同步,ETL功能

Canal-Adapter的消费源，可以使MQ，也可以是直接连Canal-Server，如下图所示：

适配器结构

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

启动器为 SpringBoot 项目, 支持canal-client启动的同时提供相关REST管理接口, 运行目录结构为:

- bin
    restart.sh
    startup.bat
    startup.sh
    stop.sh
- lib
   ...
- plugin 
    client-adapter.logger-1.1.1-jar-with-dependencies.jar
    client-adapter.hbase-1.1.1-jar-with-dependencies.jar
    ...
- conf
    application.yml
    - hbase
        mytest_person2.yml
- logs

以上目录结构最终会打包成 canal-adapter-*.tar.gz 压缩包

适配器配置说明总配置文件 application.yml

adapter 定义部分

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                        # 日志打印适配器
......

说明:

一份数据可以被多个group同时消费, 多个group之间会是一个并行执行, 一个group内部是一个串行执行多个outerAdapters, 比如例子中logger和hbase
目前client adapter数据订阅的方式支持两种，直连canal server 或者订阅kafka/RocketMQ的消息

使用远程配置（MySQL）

参考官网：https://github.com/alibaba/canal/wiki/ClientAdapter

适配器启动

解压 canal.adapter-1.1.5.tar.gz，修改 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: 0
  timeout:
  accessKey:
  secretKey:
  consumerProperties:
    # canal tcp consumer
    canal.tcp.server.host: 192.168.25.129:11111
    canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:
  canalAdapters:
  - instance: example1 # canal instance Name or mq topic name
    groups:
    - groupId: g1
      outerAdapters:
      - name: logger

启动：

./bin/startup.sh

查看日志（logs/adapter/adapter.log），显示启动成功：

修改一笔数据：

update user set age = 30 where id = 1

重新查看上面的日志，输出了变更的数据：

其他类型的适配器，参考官方说明：https://github.com/alibaba/canal/wiki/ClientAdapter

canal应用四：canal-adapter的使用

Java相关栏目本月热门文章