一、canal安装
下载地址：https://github.com/alibaba/canal/releases
image.png
二、修改配置文件,监听数据库
默认是在example文件夹，可以修改example文件夹中的instance.properties文件配置。当然也可以根据自己的业务在conf路径下新建一个文件夹，比如：test,并将example文件夹中的instance.properties文件复制到test文件夹下。
image.png
image.png
# 找到配置文件此处位置，设置成自己要监听的数据库地址
canal.instance.master.address=127.0.0.1:3306
# 找到配置文件此处位置，配置数据库用户名、密码、及数据库名称。
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
canal.instance.defaultDatabaseName = test # test是数据库名
canal.instance.enableDruid=false
# 找到此处位置，默认是监听数据库下所有的表，可以改成监听具体的数据表。
#canal.instance.filter.regex=.*\\..*
canal.instance.filter.regex=test.test_table # test是数据库名 test_table是数据库中表的名称，多个用逗号隔开(test.table1,test.table2)
配置好后，需要修改数据库的相应配置,修改数据库配置文件my.cnf
log-bin=mysql-bin #添加这一行就ok
binlog-format=ROW #选择row模式
server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复
配置完成后，重新启动数据库，并查看；默认是不开启的，为OFF
image.png
配置canal用户
CREATE USER canal IDENTIFIED BY 'canal'; # 创建“canal”用户
GRANT SELECT,REPLICATION SLAVE,REPLICATION CLIENT ON.TO 'canal'@'%'; # 为“canal”用户赋予相应权限
FLUSH PRIVILEGES; # 刷新权限
配置完成，重启以下mysql服务即可。
三、配置rocketMQ
配置rocketMQ，将动态监听数据库发送的数据变化，推送至mq中，并对相应业务进行处理。
修改/conf/canal.properties文件
# 找到以下配置，默认的是example文件夹，由于根据业务建立了test文件夹，所以需要改成test
# canal.destinations = example
canal.destinations = test # 可以配置多个，用逗号隔开，如：example,test
# 配置rocketMQ
rocketmq.producer.group = test-group # 根据自己的业务进行修改
rocketmq.enable.message.trace = false
rocketmq.customized.trace.topic = test_canal_topic # 设置topic
rocketmq.namespace =
rocketmq.namesrv.addr = 127.0.0.1:9876 # 配置rocketMQ地址
rocketmq.retry.times.when.send.failed = 0
rocketmq.vip.channel.enabled = false
rocketmq.tag =
配置好rocketMQ后，还需要修改/conf/test/instance.properties文件的topic配置
# 配置mq中的topic
canal.mq.topic=canal_test_topic
四、启动canal
启动/bin/startup.bat即可。
五、springboot项目监听canal发送rocketMQ的代码
springboot中的rocketMQ配置
#rocketmq配置
rocketmq:
name-server: 127.0.0.1:9876
producer:
group: test-group #这个需要和canal中/conf/canal.properties文件中的rocketmq.producer.group配置一致
springboot相应代码
# topic 需要和canal中的/conf/test/instance.properties文件中的canal.mq.topic配置一致
# consumerGroup = "consumer-test-canal-group" # 自定义一个消费组即可
# customizedTraceTopic需要和canal中的/conf/canal.properties文件中的rocketmq.customized.trace.topic配置一致
@Component
@RocketMQMessageListener(topic = "canal_test_topic", consumerGroup = "consumer-test-canal-group", customizedTraceTopic = "test_canal_topic")
public class CanalConsumer implements RocketMQListener<String> {
@Override
public void onMessage(String msg) {
System.out.println(msg);
if (!JSONUtil.isJson(msg)) {
return;
}
JSONObject msgJsonObject = JSONUtil.parseObj(msg);
String sqlType = msgJsonObject.getStr("type");
switch (sqlType) {
case "UPDATE":
case "INSERT":
case "DELETE":
break;
default:
System.out.println("不同步的消息类型：" + sqlType);
return;
}
System.out.println(msgJsonObject.toStringPretty());
String table = msgJsonObject.getStr("table"); # 获取表名
JSONArray data = msgJsonObject.getJSONArray("data");
JSONArray pkNames = msgJsonObject.getJSONArray("pkNames");
System.out.println("表名为:" + msgJsonObject.getStr("table") + ",sql类型为：" + sqlType);
if ("UPDATE".equals(sqlType) || "INSERT".equals(sqlType)) {
# 根据自己的业务进行处理
for (int i = 0; i < data.size(); i++) {
JSONObject object = data.getJSONObject(i);
# 可以根据序号及字段获取相应信息
String key = object.getStr(pkNames.getStr(i), "0");
String id = object.getStr("id","");
}
return;
}
if ("DELETE".equals(sqlType)) {
# 根据自己的业务进行处理
for (int i = 0; i < data.size(); i++) {
JSONObject object = data.getJSONObject(i);
String key = object.getStr(pkNames.getStr(i), "0");
}
}
}
}
