便于使用的canal-springboot客户端
还在测试使用阶段,如果在使用过程中发现问题,欢迎 提交issue 或者 加入qq群 讨论
-
环境依赖
jdk >= 1.8
canal-server = 1.1.6
-
引入maven依赖
<dependency> <groupId>io.github.xizixuejie</groupId> <artifactId>canal-spring-boot-starter</artifactId> <version>0.0.14</version> </dependency>
-
配置canal服务基本信息
canal: server: 127.0.0.1:11111 destination: example
-
在启动类添加注解
@EnableCanalListener
value
属性或者basePackages
属性指定扫描包路径。 -
自定义CanalListener 实现
io.xzxj.canal.core.listener.EntryListener
接口 -
实现类上增加注解
@CanalListener(schemaName = "${database}", tableName = "${table_name}")
${database}
监听数据库名${table_name}
监听表名。可以是正则表达式接口泛型是对应的实体类
-
实现
insert
、update
或者delete
方法来监听你想做的操作
@CanalListener
public class TestListener implements EntryListener<TestEntity> {
private static final Logger log = LoggerFactory.getLogger(TestListener.class);
@Override
public void insert(TestEntity testEntity) {
log.info("insert={}", testEntity);
}
@Override
public void update(TestEntity before, TestEntity after) {
log.info("update before={}", before);
log.info("update after={}", after);
}
@Override
public void delete(TestEntity testEntity) {
log.info("delete={}", testEntity);
}
}
打勾的是已实现的,未打勾的是以后会实现的
- 实体类名自动转换表名
- 跟据jpa或者mp注解自动转换实体类属性,或者其他orm框架的注解(需要自定义)
- tcp模式
- kafka模式
- rocketMQ模式
- rabbitMQ模式
- pulsarMQ模式
如果你用的orm框架不是mybatis-plus或者spring-data-jpa,可以自定义一个转换器。
- 自定义bean 继承
io.xzxj.canal.core.metadata.AbstractEntityInfoHelper
- 实现抽象类中的三个方法
- 获取数据库表名
getTableName
- 是否为数据库列
isColumnFiled
- 获取对应的数据库列名
getColumn
- 获取数据库表名
如果你的实体类属性不在 io.xzxj.canal.core.convertor.impl
这个包里实现,可以自定义类型转换器。
全局配置方式
- 自定义bean实现
io.xzxj.canal.core.convertor.IColumnConvertor<T>
,泛型是你实体类的类型。 - 实现
convert
方法
注解单独指定
优先级比全局高
- 在实体类属性上添加
@ColumnConvertor
注解 - value是
io.xzxj.canal.core.convertor.IColumnConvertor<T>
的实现类,需要有一个无参构造函数。
只在kafka模式下生效
可以加在你的 EntryListener<T>
的实现类上,用来指定监听具体的哪个topic以及partition,如果只指定了topic,则会监听这个topic的所有partition。
属性 | 描述 | 默认值 |
---|---|---|
canal.server | canal服务地址 如果是Kafka模式为Kafka地址,多个地址以 , 分隔。如果是rabbitMQ模式为rabbitMQ服务地址,目前只支持单个 |
127.0.0.1:11111 |
canal.annotation-type | 选择实体类上的注解类型 | mybatis_plus |
canal.destination | canal 的instance 名称 rabbitMQ模式为queue名称 |
example |
canal.filter | canal过滤的表名称,如配置则只订阅配置的表 | "" |
canal.async | 是否是异步消费,异步消费时,消费时异常将导致消息不会回滚,也不保证顺序性 | true |
canal.timeout | 消费的时间间隔(s) | 1s |
canal.server-mode | canal 客户端类型,目前支持 tcp,kafka,rabbitMQ类型 | tcp |
canal.username | canal 的用户名 | null |
canal.password | canal 的密码 | null |
canal.mq.flat-message | JSON 消息格式 | true |
canal.kafka.group-id | kafka groupId 消费者订阅消息时可使用,kafka canal 客户端 | null |
canal.kafka.dynamic-topic | kafka消费者订阅消息的topic和partition,如果用这个属性指定topic配置,则会忽略canal.kafka.topics配置 | {} |
canal.kafka.topics | kafka消费者订阅消息的topic | [] |
canal.kafka.partition | kafka partition 已过时,如果需要设置topic分区,请使用dynamicTopic来指定 | null |
canal.rabbit-mq.virtual-host | rabbitMq virtualHost | / |
canal.rabbit-mq.queue-name | rabbitMq 队列名 | null |
canal.rabbit-mq.access-key | 阿里的ak | "" |
canal.rabbit-mq.secret-key | 阿里的sk | "" |
canal.rabbit-mq.resource-owner-id | 资源owner账号(主账号 | null |
canal.rabbit-mq.username | rabbitMq username | guest |
canal.rabbit-mq.password | rabbitMq password | guest |
- 2024-03-27 v0.0.14 修改了多个destination时,CanalClient销毁出现异常的问题
- 2024-03-15 v0.0.13 优化了异常处理;修改了spring自动配置
- 2024-03-12 v0.0.12 修改了tableName正则配置失效问题
- 2024-03-12 v0.0.11 修改了一些bug;kafka模式和tcp模式支持多个destination,而且在kafka模式下支持自定义topic、partition
- 2023-12-26 v0.0.10 修改了一些bug,支持一个表对应多个EntryListener,支持注解指定自定义实体类属性类型转换。
- 2023-12-18 v0.0.9 允许自定义实体类属性名称和类型转换。
- 2023-12-13 v0.0.8 优化Listener处理过程;mq模式配置flat-message。
- 2023-09-25 v0.0.7 优化定义多个同库不同表Listener导致只能获取到最后一个Listener信息的问题。
- 2023-09-01 v0.0.6 增加List字段转换,修改了默认tcp模式自动配置不生效的问题。
- 2023-08-16 更新v0.0.5版本。
@CanalListener
注解的schemaName
属性可以指定多个,也可以使用正则表达。 - 2023-05-31 更新v0.0.4版本。修改消费消息异常处理。
- 2023-05-04 更新v0.0.3版本。
@CanalListener
注解的tableName
属性可以用正则表达式。 - 2023-03-25 更新v0.0.2版本。实现了RabbitMQ类型客户端。
- 2023-03-22 更新v0.0.1版本。实现指定库名、表名监听,实体类属性自动转换,tcp和kafka类型客户端。