Spring Boot整合阿里开源中间件Canal实现数据增量同步

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

# url
canal.instance.master.address=127.0.0.1:3306
# username/password
canal.instance.dbUsername=root
canal.instance.dbPassword=root
# 监听的数据库
canal.instance.defaultDatabaseName=test
# 监听的表，可以指定，多个用逗号分割，这里正则是监听所有
canal.instance.filter.regex=.*\\..*

# 传输方式：tcp, kafka, rocketMQ, rabbitMQ
canal.serverMode = rabbitMQ
##################################################
######### 		    RabbitMQ	     #############
##################################################
rabbitmq.host = 127.0.0.1
rabbitmq.virtual.host =/
# exchange
rabbitmq.exchange =canal.exchange
# 用户名、密码
rabbitmq.username =guest
rabbitmq.password =guest
## 是否持久化
rabbitmq.deliveryMode = 2

canal.mq.topic=canal.routing.key

INSERT INTO `oauth_client_details` VALUES ('myjszl', 'res1', '$2a$10$F1tQdeb0SEMdtjlO8X/0wO6Gqybu6vPC/Xg8OmP9/TL1i4beXdK9W', 'all', 'password,refresh_token,authorization_code,client_credentials,implicit', 'http://www.baidu.com', NULL, 1000, 1000, NULL, 'false');

{
	"data": [{
		"client_id": "myjszl",
		"resource_ids": "res1",
		"client_secret": "$2a$10$F1tQdeb0SEMdtjlO8X/0wO6Gqybu6vPC/Xg8OmP9/TL1i4beXdK9W",
		"scope": "all",
		"authorized_grant_types": "password,refresh_token,authorization_code,client_credentials,implicit",
		"web_server_redirect_uri": "http://www.baidu.com",
		"authorities": null,
		"access_token_validity": "1000",
		"refresh_token_validity": "1000",
		"additional_information": null,
		"autoapprove": "false"
	}],
	"database": "test",
	"es": 1640337532000,
	"id": 7,
	"isDdl": false,
	"mysqlType": {
		"client_id": "varchar(48)",
		"resource_ids": "varchar(256)",
		"client_secret": "varchar(256)",
		"scope": "varchar(256)",
		"authorized_grant_types": "varchar(256)",
		"web_server_redirect_uri": "varchar(256)",
		"authorities": "varchar(256)",
		"access_token_validity": "int(11)",
		"refresh_token_validity": "int(11)",
		"additional_information": "varchar(4096)",
		"autoapprove": "varchar(256)"
	},
	"old": null,
	"pkNames": ["client_id"],
	"sql": "",
	"sqlType": {
		"client_id": 12,
		"resource_ids": 12,
		"client_secret": 12,
		"scope": 12,
		"authorized_grant_types": 12,
		"web_server_redirect_uri": 12,
		"authorities": 12,
		"access_token_validity": 4,
		"refresh_token_validity": 4,
		"additional_information": 12,
		"autoapprove": 12
	},
	"table": "oauth_client_details",
	"ts": 1640337532520,
	"type": "INSERT"
}

/**
 * @author 公号 码猿技术专栏
 * Canal消息接收实体类
 */
@NoArgsConstructor
@Data
public class CanalMessage<T> {
    @JsonProperty("type")
    private String type;
    @JsonProperty("table")
    private String table;
    @JsonProperty("data")
    private List<T> data;
    @JsonProperty("database")
    private String database;
    @JsonProperty("es")
    private Long es;
    @JsonProperty("id")
    private Integer id;
    @JsonProperty("isDdl")
    private Boolean isDdl;
    @JsonProperty("old")
    private List<T> old;
    @JsonProperty("pkNames")
    private List<String> pkNames;
    @JsonProperty("sql")
    private String sql;
    @JsonProperty("ts")
    private Long ts;
}

import cn.hutool.json.JSONUtil;
import cn.myjszl.middle.ware.canal.mq.rabbit.model.CanalMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
 * 监听MQ获取Canal增量的数据消息
 */
@Component
@Slf4j
@RequiredArgsConstructor
public class CanalRabbitMQListener {
    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue(value = "canal.queue", durable = "true"),
                    exchange = @Exchange(value = "canal.exchange"),
                    key = "canal.routing.key"
            )
    })
    public void handleDataChange(String message) {
        //将message转换为CanalMessage
        CanalMessage canalMessage = JSONUtil.toBean(message, CanalMessage.class);
        String tableName = canalMessage.getTable();
        log.info("Canal 监听 {} 发生变化；明细：{}", tableName, message);
        //TODO 业务逻辑自己完善...............
    }
}

INSERT INTO `oauth_client_details`
VALUES
	( 'myjszl', 'res1', '$2a$10$F1tQdeb0SEMdtjlO8X/0wO6Gqybu6vPC/Xg8OmP9/TL1i4beXdK9W', 'all', 'password,refresh_token,authorization_code,client_credentials,implicit', 'http://www.baidu.com', NULL, 1000, 1000, NULL, 'false' );