MySQL与Canal、RabbitMQ集成指南

MySQL 部分

1. 查看是否开启 binlog

MySQL 8 默认开启 binlog。可以通过以下命令查看是否开启：

SHOW VARIABLES LIKE 'log_bin';

如果返回结果为 ON，则表示 binlog 已开启。

Variable_name	Value
log_bin	ON

2. 若未开启 binlog，则需手动配置

如果 binlog 未开启，需要在 MySQL 配置文件中添加以下配置：

log-bin=mysql-bin  # 开启 binlog
server_id=1        # 配置 MySQL replication 需要定义，确保不与 Canal 的 slaveId 重复

修改完成后，重启 MySQL 使配置生效。

3. 创建 Canal 使用的 MySQL 用户

Canal 需要连接到 MySQL 并读取 binlog，因此需要创建一个专门的用户并授予相应权限。

# 创建用户
CREATE USER canal IDENTIFIED WITH mysql_native_password BY 'canal';# 授予权限
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';# 刷新权限
FLUSH PRIVILEGES;

MQ 部分

在 RabbitMQ 中，我们需要创建一个交换机和队列，并将它们绑定在一起。我使用的是已经创建过的 Virtual Host ： trovebox_dev，你可以根据实际情况决定是否创建新的 Virtual Host。

1. 新建交换机

在 RabbitMQ 管理界面中，创建一个新的交换机，命名为 canal.exchange。

新建交换机

2. 添加队列

创建一个新的队列，命名为 canal.queue。

添加队列

3. 绑定交换机

将队列 canal.queue 绑定到交换机 canal.exchange，并设置路由键为 canal.routing.key。

绑定交换机

Canal 部分

Docker 安装 Canal

使用 Docker 安装 Canal 非常简单，以下是安装步骤：

拉取 Canal 镜像：没有tag默认最新的
```
docker pull canal/canal-server
```

运行 Canal 容器：

docker run -p 11111:11111 -p 11110:11110 -p 11112:11112 \--name canal \-e canal.destinations=destination \-e canal.instance.master.address=ip:port \-e canal.instance.dbUsername=canal \-e canal.instance.dbPassword=canal \-e canal.instance.connectionCharset=UTF-8 \-e canal.instance.tsdb.enable=true \-e canal.instance.gtidon=false \-e canal.instance.filter.regex=dataBaseName\\..* \-d canal/canal-server:latest

将 Canal 的配置文件和日志文件拷贝到宿主机：

docker cp containerId:/home/admin/canal-server/conf /www/dk_project/dk_app/canal/
docker cp containerId:/home/admin/canal-server/logs /www/dk_project/dk_app/canal/

修改配置文件 conf/canal.properties：

canal.serverMode = rabbitMQ
rabbitmq.host = ip
rabbitmq.virtual.host = trovebox_dev
rabbitmq.exchange = canal.exchange
rabbitmq.username = trovebox_dev
rabbitmq.password = troveboxadmin

修改配置文件 conf/destination/canal.properties：

canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.mq.topic=canal.routing.key

删除并重新创建 Canal 容器：

docker rm -f canaldocker run -p 11111:11111 -p 11110:11110 -p 11112:11112 \--name canal \-e canal.destinations=destination \-e canal.instance.master.address=ip:port \-e canal.instance.dbUsername=canal \-e canal.instance.dbPassword=canal \-e canal.instance.connectionCharset=UTF-8 \-e canal.instance.tsdb.enable=true \-v /www/dk_project/dk_app/canal/conf:/home/admin/canal-server/conf/ \-v /home/admin/canal-server/logs:/home/admin/canal-server/logs/ \-e canal.instance.filter.regex=dataBaseName\\..* \-d canal/canal-server:latest

Java 部分代码

BinLogDto.java

BinLogDto 类用于解析 Canal 发送的 binlog 数据。

package online.trovebox.ruyiai.common.dto;import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import lombok.Data;import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;@Data
public class BinLogDto {private String database;  // 数据库private String table;     // 表private String type;      // 操作类型private JSONArray data;   // 操作数据private JSONArray old;    // 变更前数据private JSONArray pkNames; // 主键名称private String sql;       // 执行 SQL 语句private Long es;private String gtid;private Long id;private Boolean isDdl;private JSONObject mysqlType;private JSONObject sqlType;private Long ts;public <T> List<T> getData(Class<T> clazz) {if (this.data == null || this.data.size() == 0) {return null;}return this.data.toList(clazz);}public <T> List<T> getOld(Class<T> clazz) {if (this.old == null || this.old.size() == 0) {return null;}return this.old.toList(clazz);}public List<String> getPkNames() {if (this.pkNames == null || this.pkNames.size() == 0) {return null;}List<String> pkNames = new ArrayList<>();for (Object pkName : this.pkNames) {pkNames.add(pkName.toString());}return pkNames;}public Map<String, String> getMysqlType() {if (this.mysqlType == null) {return null;}Map<String, String> mysqlTypeMap = new HashMap<>();this.mysqlType.forEach((k, v) -> {mysqlTypeMap.put(k, v.toString());});return mysqlTypeMap;}public Map<String, Integer> getSqlType() {if (this.sqlType == null) {return null;}Map<String, Integer> sqlTypeMap = new HashMap<>();this.sqlType.forEach((k, v) -> {sqlTypeMap.put(k, Integer.valueOf(v.toString()));});return sqlTypeMap;}
}

Listener.java

Listener 类用于监听 RabbitMQ 中的消息，并处理 binlog 数据。

package online.trovebox.ruyiai.listener;import com.alibaba.fastjson2.JSON;
import jakarta.annotation.Resource;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import online.trovebox.ruyiai.common.dto.BinLogDto;
import org.springframework.amqp.core.Message;
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.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;import java.nio.charset.StandardCharsets;
import java.util.HashSet;
import java.util.Set;@Component
@Slf4j
@RequiredArgsConstructor
public class CanalListener {@Resourceprivate RedisTemplate<String, Object> redisTemplate;String[] prefixes = new String[]{"coin_change_log","log","message",};@RabbitListener(bindings = {@QueueBinding(value = @Queue(value = "canal.queue", durable = "true"),exchange = @Exchange(value = "canal.exchange"),key = "canal.routing.key")})public void handleDataChange(@Payload Message message) {String content = new String(message.getBody(), StandardCharsets.UTF_8);BinLogDto binLog = JSON.parseObject(content, BinLogDto.class);String type = binLog.getType();if (type.equalsIgnoreCase("select")) {return;}String table = binLog.getTable();for (String prefix : prefixes) {if (table.startsWith(prefix)) {System.err.println(table);return;}}log.info("表：{} 操作类型:{}", table, binLog.getType());log.info("操作后数据：{} ", binLog.getData().toStringPretty());deleteKeysStartingWith(table);}public void deleteKeysStartingWith(String prefix) {String cursor = "0";do {Set<String> keys = scanKeys(cursor, prefix);cursor = keys.isEmpty() ? "0" : "1";if (!keys.isEmpty()) {redisTemplate.delete(keys);}} while (!"0".equals(cursor));}private Set<String> scanKeys(String cursor, String prefix) {return redisTemplate.execute((RedisCallback<Set<String>>) connection -> {ScanOptions options = ScanOptions.scanOptions().match(prefix + "*").count(1000).build();Cursor<byte[]> cursorScan = connection.scan(options);Set<String> keys = new HashSet<>();while (cursorScan.hasNext()) {byte[] keyBytes = cursorScan.next();keys.add(new String(keyBytes, StandardCharsets.UTF_8));}return keys;});}
}