/canal-client

spring boot canal starter 易用的canal 客户端 canal client

Primary LanguageJava

Build Status

易用的canal 客户端 easy canal client

介绍

canal 是阿里巴巴mysql数据库binlog的增量订阅&消费组件
使用该客户端前请先了解canal,https://github.com/alibaba/canal
canal 自身提供了简单的客户端，如果要转换为数据库的实体对象，处理消费数据要每次进行对象转换。 该客户端直接将canal的数据原始类型转换为各个数据表的实体对象，并解耦数据的增删改操作，方便给业务使用。

要求

java8+

特性

  • 解耦单表增删操作
  • simple,cluster,zookeeper,kafka客户端支持
  • 同步异步处理支持
  • spring boot 开箱即用

如何使用

spring boot 方式 maven 依赖

  <dependency>
      <groupId>top.javatool</groupId>
      <artifactId>canal-spring-boot-starter</artifactId>
      <version>1.2.6-RELEASE</version>
  </dependency>

java 方式

 <dependency>
       <groupId>top.javatool</groupId>
       <artifactId>canal-client</artifactId>
       <version>1.2.6-RELEASE</version>
 </dependency>

配置说明

属性 描述 默认值
canal.mode canal 客户端类型 目前支持4种类型 simple,cluster,zk,kafka(kafka 目前支持flatMessage 格式) simple
canal.filter canal过滤的表名称，如配置则只订阅配置的表 ""
canal.batch-size 消息的数量，超过该次数将进行一次消费 1(个)
canal.timeout 消费的时间间隔(s) 1s
canal.server 服务地址,多个地址以,分隔 格式 ${host}:${port} null
canal.destination canal 的instance 名称,kafka模式为topic 名称 null
canal.user-name canal 的用户名 null
canal.password canal 的密码 null
canal.group-id kafka groupId 消费者订阅消息时可使用，kafka canal 客户端 null
canal.async 是否是异步消费，异步消费时，消费时异常将导致消息不会回滚，也不保证顺序性 true
canal.partition kafka partition null

订阅数据库的增删改操作

实现EntryHandler 接口，泛型为想要订阅的数据库表的实体对象， 该接口的方法为 java 8 的 default 方法，方法可以不实现，如果只要监听增加操作，只实现增加方法即可
下面以一个t_user表的user实体对象为例, 默认情况下，将使用实体对象的jpa 注解 @Table中的表名来转换为EntryHandler中的泛型对象，

public class UserHandler implements EntryHandler<User>{

}

如果实体类没有使用jpa @Table的注解，也可以使用@CanalTable 注解在EntryHandler来标记表名，例如

@CanalTable(value = "t_user")
@Component
public class UserHandler implements EntryHandler<User>{

   /**
   *  新增操作
   * @param user
    */
   @Override
    public void insert(User user) {
	   //你的逻辑
        log.info("新增 {}",user);
    }
    /**
    * 对于更新操作来讲，before 中的属性只包含变更的属性，after 包含所有属性，通过对比可发现那些属性更新了
   * @param before
   * @param after
    */
    @Override
    public void update(User before, User after) {
    	//你的逻辑
        log.info("更新 {} {}",before,after);
    }
    /**
    *  删除操作
    * @param user
    */
    @Override
    public void delete(User user) {
       //你的逻辑
        log.info("删除 {}",user); 
   }
}

另外也支持统一的处理@CanalTable(value="all"),这样除去存在EntryHandler的表以外，其他所有表的处理将通过该处理器,统一转为Map<String, String>对象

@CanalTable(value = "all")
@Component
public class DefaultEntryHandler implements EntryHandler<Map<String, String>> {
     @Override
        public void insert(Map<String, String> map) {
            logger.info("insert message  {}", map);
        }
    
        @Override
        public void update(Map<String, String> before, Map<String, String> after) {
            logger.info("update before {} ", before);
            logger.info("update after {}", after);
        }
    
        @Override
        public void delete(Map<String, String> map) {
            logger.info("delete  {}", map);
        }
}

如果你想获取除实体类信息外的其他信息，可以使用

CanalModel canal = CanalContext.getModel();

具体使用可以查询项目demo 示例
https://github.com/NormanGyllenhaal/canal-client/tree/master/canal-example