建站之星源码下载小程序打不开什么原因
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;});}
}
效果图
知识点说明
- Binlog：MySQL 的二进制日志，用于记录数据库的所有更改操作。Canal 通过读取 binlog 来获取数据库的变更数据。
- Canal：阿里巴巴开源的数据库同步工具，基于 MySQL 的 binlog 实现数据同步。
- RabbitMQ：消息队列中间件，用于在分布式系统中传递消息。Canal 可以将 binlog 数据发送到 RabbitMQ，供其他服务消费。
- Redis：内存数据库，用于缓存数据。在监听 binlog 变更时，可以通过 Redis 缓存相关数据，并在数据变更时清除缓存。
通过以上步骤和代码，你可以实现 MySQL 数据库的变更监听，并将变更数据通过 RabbitMQ 发送到其他服务进行处理。