微服务架构-分布式解决方案-109：基于canal框架解决mysql与redis一致性的问题
canal
alibaba/canal: Canal 是由阿里巴巴开源的分布式数据库同步系统，主要用于实现MySQL数据库的日志解析和实时增量数据订阅与消费，广泛应用于数据库变更消息的捕获、数据迁移、缓存更新等场景。
项目地址：https://gitcode.com/gh_mirrors/ca/canal
免费下载资源
·
109：基于canal框架解决mysql与redis一致性的问题
1 canal解决MySql与Redis一致性问题演示
课题内容
- MySQL与Redis一致性的问题如何解决
- 每特教育mt_sqlsyn同步框架原理
- 阿里巴巴canal的架构设计原理
- canal配置实例演示
2 redis与Mysql不一致性的问题如何解决
mysql与redis一致性的实现原理–最终一致性
Mysql与Redis不一致性的情况下如何解决？
- 直接将对应的key在Redis缓存中删除掉，Redis缓存没有会重新查询mysql并将将该内容同步到Redis，这种方式效率非常低；
- 定时刷新Redis缓存，延迟的概率太大；
3 每特教育sqlsyn同步框架原理
- 使用每特教育mt_sqlsyn同步框架。
原理：直接代理项目中数据源，拦截inset、update、delete语句。如果本地事务commit成功之后，异步的形式（多线程或者MQ）将该数据同步到Redis中。
技术栈：线程池、并发队列、MQ、设计模式思想、jvm、性能测试、整合SpringBoot。
缺陷：人工修改mysql数据库无法自动同步，有提供api刷新接口。
4 阿里巴巴canal基本的介绍
- 阿里巴巴的canal框架
canal，译意为水道/管道/沟渠，主要用途是基于MySQL数据库增量日志解析，提供增量数据订阅和消费。
基于日志增量订阅和消费的业务包括
数据库镜像、数据库实时备份、索引构建和实时维护(拆分异构索引、倒排索引等)、业务 cache 刷新、带业务逻辑的增量数据处理
当前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x
5 MySQL主从复制实现的原理
主：读和写 从：只能读不能写
主从复制的原理：从节点订阅主的mysql服务器，主节点binlog文件发生变化的情况下，将该binlog文件增量的形式同步给从节点服务器，这样的话主和从能够最终保持数据的一致性。
主从复制作用：
1.对数据实现备份
2.读写分离
3.故障转移
4.实现简单集群
6 阿里巴巴canal的实现原理分析
canal会启动一个Server端作为一个mysql从节点拉取mysql主节点最新的binlog文件，只要mysql主节点的binlog文件发生变化都会以增量的形式通知给canalServer端，canalServer再通知给canalClient，canalClient自己手动配置刷新Redis接口。
7 开启MySQL主从复制的权限
1.配置Mysql服务器
配置MySQL的my.ini/my.cnf(linux)开启允许基于binlog文件主从同步
log-bin=mysql-bin #添加这一行就ok
binlog-format=ROW #选择row模式
server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复
配置该文件后，重启mysql服务器即可
show variables like ‘log_bin’;
没有开启log_bin的值是OFF,开启之后是ON
2.添加cannl的账号
drop user 'canal'@'%';
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
grant all privileges on *.* to 'canal'@'%' identified by 'canal';
flush privileges;
select * from `mysql`.`user` where user='canal';
检查mysql user表canal账号权限为Y
8 启动canal服务器端服务
构建CanalService
修改\conf\example下的instance.properties 配置文件内容
canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
启动startup.bat，查看\logs\example example.log日志文件
9 canalClient连接canal服务器
引入maven依赖
<dependencies>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.0</version>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>
</dependencies>
同步代码
package com.mayikt;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import java.net.InetSocketAddress;
import java.util.List;
public class CanalClient {
public static void main(String args[]) {
// 连接canalServer端
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1",
11111), "example", "", "");
int batchSize = 100;
try {
connector.connect();
// 只监听test库的users表
connector.subscribe("test.user");
connector.rollback();
while (true) {
// 获取指定数量的数据
Message message = connector.getWithoutAck(batchSize);
long batchId = message.getId();
int size = message.getEntries().size();
System.out.println("batchId = " + batchId);
System.out.println("size = " + size);
if (batchId == -1 || size == 0) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
printEntry(message.getEntries());
}
// 提交确认
connector.ack(batchId);
// connector.rollback(batchId); // 处理失败, 回滚数据
}
} finally {
connector.disconnect();
}
}
private static void printEntry(List<Entry> entrys) {
for (Entry entry : entrys) {
if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
continue;
}
RowChange rowChage = null;
try {
rowChage = RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
e);
}
EventType eventType = rowChage.getEventType();
System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
eventType));
for (RowData rowData : rowChage.getRowDatasList()) {
if (eventType == EventType.DELETE) {
redisDelete(rowData.getBeforeColumnsList());
} else if (eventType == EventType.INSERT) {
redisInsert(rowData.getAfterColumnsList());
} else {
System.out.println("-------> before");
printColumn(rowData.getBeforeColumnsList());
System.out.println("-------> after");
printColumn(rowData.getAfterColumnsList());
redisUpdate(rowData.getAfterColumnsList());
}
}
}
}
private static void printColumn(List<Column> columns) {
for (Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
private static void redisInsert(List<Column> columns) {
JSONObject json = new JSONObject();
for (Column column : columns) {
json.put(column.getName(), column.getValue());
}
if (columns.size() > 0) {
RedisUtil.stringSet(columns.get(0).getValue(), json.toJSONString());
}
}
private static void redisUpdate(List<Column> columns) {
JSONObject json = new JSONObject();
for (Column column : columns) {
json.put(column.getName(), column.getValue());
}
if (columns.size() > 0) {
RedisUtil.stringSet(columns.get(0).getValue(), json.toJSONString());
}
}
private static void redisDelete(List<Column> columns) {
JSONObject json = new JSONObject();
for (Column column : columns) {
json.put(column.getName(), column.getValue());
}
if (columns.size() > 0) {
RedisUtil.delKey(columns.get(0).getValue());
}
}
}
package com.mayikt;
import redis.clients.jedis.Jedis;
public class RedisUtil {
private static Jedis jedis = null;
public static synchronized Jedis getJedis() {
if (jedis == null) {
jedis = new Jedis("127.0.0.1", 6379);
}
return jedis;
}
public static boolean existKey(String key) {
return getJedis().exists(key);
}
public static void delKey(String key) {
getJedis().del(key);
}
public static String stringGet(String key) {
return getJedis().get(key);
}
public static String stringSet(String key, String value) {
return getJedis().set(key, value);
}
public static void hashSet(String key, String field, String value) {
getJedis().hset(key, field, value);
}
}
测试效果：
GitHub 加速计划 / ca / canal
28.22 K
7.57 K
下载
alibaba/canal: Canal 是由阿里巴巴开源的分布式数据库同步系统，主要用于实现MySQL数据库的日志解析和实时增量数据订阅与消费，广泛应用于数据库变更消息的捕获、数据迁移、缓存更新等场景。
最近提交(Master分支：3 个月前 )
1e5b8a20 - 2 个月前
ff82fd65
2 个月前
旨在为数千万中国开发者提供一个无缝且高效的云端环境，以支持学习、使用和贡献开源项目。
更多推荐
0
0- 0
已为社区贡献3条内容
所有评论(0)