背景
项目架构SpringMVC+MyBatisPlus+Redis。
开发过程中遇到的问题:
Redis和MySQL会出现数据不一致问题,因此使用canal中间件来操作Redis中的数据(增加、删除、修改)。
使用canal之后,业务服务只管从Redis中取数据,不必关心Redis中的数据从何而来。
0. alibaba/canal介绍
canal 主要用途是基于 MySQL 数据库增量日志解析,提供增量数据订阅和消费。
当前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x
1. 需要MySQL开启bin_log功能
执行show variables like '%log_bin%'
查看是否启用bin_log功能
2. 部署canal-deployer
canal下载地址
canal-deployer-1.1.5.tar.gz下载地址
教程参考
3. 开发canal客户端
用于接收canal-deployer发来的bin_log数据
pom.xml
<repositories>
<repository>
<id>alimaven</id>
<name>aliyun maven</name>
<url>http://maven.aliyun.com/nexus/content/groups/public/</url>
</repository>
</repositories>
<properties>
<java.version>1.8</java.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>3.2.0</version>
</dependency>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.4</version>
</dependency>
</dependencies>
CanalClient.java
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[]) {
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1",
11111), "example", "canal", "canal");
int batchSize = 1000;
try {
connector.connect();
connector.subscribe(".*\\..*");
connector.rollback();
while (true) {
Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
long batchId = message.getId();
int size = message.getEntries().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(entry.getHeader().getTableName(), rowData.getBeforeColumnsList());
} else if (eventType == EventType.INSERT) {
redisInsert(entry.getHeader().getTableName(), rowData.getAfterColumnsList());
} else {
System.out.println("-------> before");
printColumn(rowData.getBeforeColumnsList());
System.out.println("-------> after");
redisUpdate(entry.getHeader().getTableName(), 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(String tableName, List<Column> columns) {
JSONObject json = new JSONObject();
for (Column column : columns) {
json.put(column.getName(), column.getValue());
}
if (columns.size() > 0) {
System.out.println(columns.get(0).getValue() + "," + json.toJSONString());
RedisUtil.stringSet(tableName, columns.get(0).getValue(), json.toJSONString());
}
}
private static void redisUpdate(String tableName, List<Column> columns) {
JSONObject json = new JSONObject();
for (Column column : columns) {
json.put(column.getName(), column.getValue());
}
if (columns.size() > 0) {
System.out.println(columns.get(0).getValue() + "," + json.toJSONString());
RedisUtil.stringSet(tableName, columns.get(0).getValue(), json.toJSONString());
}
}
private static void redisDelete(String tableName, List<Column> columns) {
JSONObject json = new JSONObject();
for (Column column : columns) {
json.put(column.getName(), column.getValue());
}
if (columns.size() > 0) {
System.out.println(columns.get(0).getValue());
RedisUtil.delKey(tableName, columns.get(0).getValue());
}
}
}
RedisUtil.java
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
/**
* @author lgz
*/
public class RedisUtil {
private static Jedis jedis = null;
public static synchronized Jedis getJedis() {
if (jedis == null) {
jedis = new Jedis("127.0.0.1", 6379);
jedis.auth("123456");
}
return jedis;
}
public static boolean existKey(String key) {
return getJedis().exists(key);
}
// todo 增删改操作需要清空所有filter数据
// getJedis().del(tableName+":filter");
public static void delKey(String tableName,String key) {
getJedis().hdel(tableName,key);
}
public static String stringGet(String tableName,String key) {
return getJedis().hget(tableName, key);
}
public static String stringSet(String tableName,String key, String value) {
getJedis().hset(tableName,key,value);
return "";
// return getJedis().set(key, value);
}
public static void hashSet(String key, String field, String value) {
getJedis().hset(key, field, value);
}
}