chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(canal/server): 优化程序代码 (#5545)

Browse Source 
- canal server logback.xml 配置增加
   CanalServerWithEmbedded 单独日志配置，流量非常大的情况下，getMessages/ack等日志量非常大
   日志刷盘太快也影响性能，占磁盘空间。用户可以根据需要单独调整日志级别。默认是INFO

- TableMetaCache 锁的范围太大，针对这块做了代码逻辑优化。以下场景影响性能：
   - 当用户关闭tsdb.enable=false, 分库分表数量非常多回查次数很多；
   - 每张表有四五百列甚至更多，里面的集合容器提前初始化，减少扩容的次数；

- DatabaseTableMeta 标记一个内存不安全的地方，以后看情况优化掉
	使用MemoryTableMeta在内存区记录表元数据，当分库分表数量非常多，表结构非常复杂非常吃内存，容易OOM
This commit is contained in:
sunxien
2026-01-28 11:52:25 +08:00
committed by GitHub
parent dcea8afc93
commit f07a075dce
3 changed files with 137 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
deployer/src/main/resources/logback.xml
View File

@@ -86,6 +86,10 @@
<level value="INFO" />
<appender-ref ref="CANAL-ROOT" />
</logger>
<logger name="com.alibaba.otter.canal.server.embedded" additivity="false">
<level value="INFO" />
<appender-ref ref="CANAL-ROOT" />
</logger>
<logger name="com.alibaba.otter.canal.meta.FileMixedMetaManager" additivity="false">
<level value="INFO" />
<appender-ref ref="CANAL-META" />

234
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java
View File

@@ -5,6 +5,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
import org.apache.commons.lang.StringUtils;
@@ -43,67 +44,83 @@ public class TableMetaCache {
private boolean isOnTSDB = false;
private TableMetaTSDB tableMetaTSDB;
// 第一层tableId,第二层schema.table,解决tableId重复，对应多张表
private LoadingCache<String, TableMeta> tableMetaDB;
private LoadingCache<String, TableMeta> tableMetaCache;
public TableMetaCache(MysqlConnection con, TableMetaTSDB tableMetaTSDB){
this.connection = con;
this.tableMetaTSDB = tableMetaTSDB;
// 如果持久存储的表结构为空，从db里面获取下
if (tableMetaTSDB == null) {
this.tableMetaDB = CacheBuilder.newBuilder().build(new CacheLoader<String, TableMeta>() {
@Override
public TableMeta load(String name) throws Exception {
try {
return getTableMetaByDB(name);
} catch (Throwable e) {
// 尝试做一次retry操作
try {
connection.reconnect();
return getTableMetaByDB(name);
} catch (IOException e1) {
throw new CanalParseException("fetch failed by table meta:" + name, e1);
}
}
}
});
this.tableMetaCache = createLocalCache();
} else {
isOnTSDB = true;
}
try {
ResultSetPacket packet = connection.query("show global variables like 'rds\\_%'");
if (packet.getFieldValues().size() > 0) {
isOnRDS = true;
}
// if (packet.getFieldValues().size() > 0) {
isOnRDS = packet.getFieldValues().size() > 0;
// }
} catch (IOException e) {
}
try {
ResultSetPacket packet = connection.query("show global variables like 'polarx\\_%'");
if (packet.getFieldValues().size() > 0) {
isOnPolarX = true;
}
// if (packet.getFieldValues().size() > 0) {
isOnPolarX = packet.getFieldValues().size() > 0;
// }
} catch (IOException e) {
}
}
private synchronized TableMeta getTableMetaByDB(String fullname) throws IOException {
try {
ResultSetPacket packet = connection.query("show create table " + fullname);
String[] names = StringUtils.split(fullname, "`.`");
String schema = names[0];
String table = names[1].substring(0, names[1].length());
return new TableMeta(schema, table, parseTableMeta(schema, table, packet));
} catch (Throwable e) { // fallback to desc table
ResultSetPacket packet = connection.query("desc " + fullname);
String[] names = StringUtils.split(fullname, "`.`");
String schema = names[0];
String table = names[1].substring(0, names[1].length());
return new TableMeta(schema, table, parseTableMetaByDesc(packet));
private LoadingCache<String, TableMeta> createLocalCache() {
return CacheBuilder.newBuilder().build(new CacheLoader<String, TableMeta>() {
@Override
public TableMeta load(String name) throws Exception {
try {
return getTableMetaByDB(name);
} catch (Throwable e) {
// 尝试做一次retry操作
try {
connection.reconnect();
return getTableMetaByDB(name);
} catch (IOException e1) {
throw new CanalParseException("fetch table meta failed. table: " + name, e1);
}
}
}
});
}
private /* synchronized */ TableMeta getTableMetaByDB(String fullname) throws IOException {
// try {
// ResultSetPacket packet = connection.query("show create table " + fullname);
// String[] names = StringUtils.split(fullname, "`.`");
// String schema = names[0];
// String table = names[1].substring(0, names[1].length());
// return new TableMeta(schema, table, parseTableMeta(schema, table, packet));
// } catch (Throwable e) { // fallback to desc table
// ResultSetPacket packet = connection.query("desc " + fullname);
// String[] names = StringUtils.split(fullname, "`.`");
// String schema = names[0];
// String table = names[1].substring(0, names[1].length());
// return new TableMeta(schema, table, parseTableMetaByDesc(packet));
// }
boolean showCreateTable = true;
ResultSetPacket packet = null;
synchronized (this) {
try {
packet = connection.query("show create table " + fullname);
}catch (Exception ex) {
packet = connection.query("desc " + fullname);
}
}
String[] names = StringUtils.split(fullname, "`.`");
String schema = names[0];
String table = names[1].substring(0, names[1].length());
List<FieldMeta> fieldMetas = showCreateTable ? parseTableMeta(schema, table, packet) : parseTableMetaByDesc(packet);
return new TableMeta(schema, table, fieldMetas);
}
public static List<FieldMeta> parseTableMeta(String schema, String table, ResultSetPacket packet) {
@@ -122,86 +139,101 @@ public class TableMetaCache {
* 处理desc table的结果
*/
public static List<FieldMeta> parseTableMetaByDesc(ResultSetPacket packet) {
Map<String, Integer> nameMaps = new HashMap<>(6, 1f);
List<FieldPacket> fieldPackets = packet.getFieldDescriptors();
int size = fieldPackets.size();
Map<String, Integer> nameMaps = new HashMap<>(size);
int index = 0;
for (FieldPacket fieldPacket : packet.getFieldDescriptors()) {
// for (FieldPacket fieldPacket : packet.getFieldDescriptors()) {
for (FieldPacket fieldPacket : fieldPackets) {
nameMaps.put(StringUtils.lowerCase(fieldPacket.getName()), index++);
}
int size = packet.getFieldDescriptors().size();
int count = packet.getFieldValues().size() / packet.getFieldDescriptors().size();
List<FieldMeta> result = new ArrayList<>();
List<String> fieldValues = packet.getFieldValues();
// int size = packet.getFieldDescriptors().size();
// int count = packet.getFieldValues().size() / packet.getFieldDescriptors().size();
int count = fieldValues.size() / size;
List<FieldMeta> result = new ArrayList<>(count);
for (int i = 0; i < count; i++) {
FieldMeta meta = new FieldMeta();
// 做一个优化，使用String.intern()，共享String对象，减少内存使用
meta.setColumnName(packet.getFieldValues().get(nameMaps.get(COLUMN_NAME) + i * size).intern());
meta.setColumnType(packet.getFieldValues().get(nameMaps.get(COLUMN_TYPE) + i * size));
meta.setNullable(StringUtils.equalsIgnoreCase(packet.getFieldValues().get(nameMaps.get(IS_NULLABLE) + i
* size),
"YES"));
meta.setKey("PRI".equalsIgnoreCase(packet.getFieldValues().get(nameMaps.get(COLUMN_KEY) + i * size)));
meta.setUnique("UNI".equalsIgnoreCase(packet.getFieldValues().get(nameMaps.get(COLUMN_KEY) + i * size)));
meta.setColumnName(fieldValues.get(nameMaps.get(COLUMN_NAME) + i * size).intern());
meta.setColumnType(fieldValues.get(nameMaps.get(COLUMN_TYPE) + i * size));
meta.setNullable(StringUtils.equalsIgnoreCase(fieldValues.get(nameMaps.get(IS_NULLABLE) + i * size), "YES"));
meta.setKey("PRI".equalsIgnoreCase(fieldValues.get(nameMaps.get(COLUMN_KEY) + i * size)));
meta.setUnique("UNI".equalsIgnoreCase(fieldValues.get(nameMaps.get(COLUMN_KEY) + i * size)));
// 特殊处理引号
meta.setDefaultValue(DruidDdlParser.unescapeQuotaName(packet.getFieldValues()
.get(nameMaps.get(COLUMN_DEFAULT) + i * size)));
meta.setExtra(packet.getFieldValues().get(nameMaps.get(EXTRA) + i * size));
meta.setDefaultValue(DruidDdlParser.unescapeQuotaName(fieldValues.get(nameMaps.get(COLUMN_DEFAULT) + i * size)));
meta.setExtra(fieldValues.get(nameMaps.get(EXTRA) + i * size));
result.add(meta);
}
return result;
}
public TableMeta getTableMeta(String schema, String table) {
return getTableMeta(schema, table, true);
}
// This method is unused
// public TableMeta getTableMeta(String schema, String table) {
// return getTableMeta(schema, table, true);
// }
public TableMeta getTableMeta(String schema, String table, boolean useCache) {
if (!useCache) {
tableMetaDB.invalidate(getFullName(schema, table));
}
// This method is unused
// public TableMeta getTableMeta(String schema, String table, boolean useCache) {
// String fullName = getFullName(schema, table);
// if (!useCache) {
// // tableMetaCache.invalidate(getFullName(schema, table));
// tableMetaCache.invalidate(fullName);
// }
// // return tableMetaCache.getUnchecked(getFullName(schema, table));
// return tableMetaCache.getUnchecked(fullName);
// }
return tableMetaDB.getUnchecked(getFullName(schema, table));
}
// This method is unused
// public TableMeta getTableMeta(String schema, String table, EntryPosition position) {
// return getTableMeta(schema, table, true, position);
// }
public TableMeta getTableMeta(String schema, String table, EntryPosition position) {
return getTableMeta(schema, table, true, position);
}
public synchronized TableMeta getTableMeta(String schema, String table, boolean useCache, EntryPosition position) {
public /* synchronized */ TableMeta getTableMeta(String schema, String table, boolean useCache, EntryPosition position) {
TableMeta tableMeta = null;
if (tableMetaTSDB != null) {
tableMeta = tableMetaTSDB.find(schema, table);
if (tableMeta == null) {
// 因为条件变化，可能第一次的tableMeta没取到，需要从db获取一次，并记录到snapshot中
String fullName = getFullName(schema, table);
ResultSetPacket packet = null;
String createDDL = null;
try {
try {
packet = connection.query("show create table " + fullName);
} catch (Exception e) {
// 尝试做一次retry操作
connection.reconnect();
packet = connection.query("show create table " + fullName);
synchronized (this) {
tableMeta = tableMetaTSDB.find(schema, table);
if (tableMeta != null) {
return tableMeta;
}
ResultSetPacket packet = null;
try {
packet = connection.query("show create table " + fullName);
} catch (Exception e) {
// 尝试做一次retry操作
connection.reconnect();
packet = connection.query("show create table " + fullName);
}
String createDDL = null;
if (packet.getFieldValues().size() > 0) {
createDDL = packet.getFieldValues().get(1);
}
// 强制覆盖掉内存值
tableMetaTSDB.apply(position, schema, createDDL, "first");
tableMeta = tableMetaTSDB.find(schema, table);
}
if (packet.getFieldValues().size() > 0) {
createDDL = packet.getFieldValues().get(1);
}
// 强制覆盖掉内存值
tableMetaTSDB.apply(position, schema, createDDL, "first");
tableMeta = tableMetaTSDB.find(schema, table);
} catch (IOException e) {
throw new CanalParseException("fetch failed by table meta:" + fullName, e);
}
}
return tableMeta;
} else {
String fullName = getFullName(schema, table);
if (!useCache) {
tableMetaDB.invalidate(getFullName(schema, table));
// tableMetaCache.invalidate(getFullName(schema, table));
tableMetaCache.invalidate(fullName);
}
return tableMetaDB.getUnchecked(getFullName(schema, table));
// return tableMetaCache.getUnchecked(getFullName(schema, table));
return tableMetaCache.getUnchecked(fullName);
}
}
@@ -209,7 +241,7 @@ public class TableMetaCache {
if (tableMetaTSDB != null) {
// tsdb不需要做,会基于ddl sql自动清理
} else {
tableMetaDB.invalidate(getFullName(schema, table));
tableMetaCache.invalidate(getFullName(schema, table));
}
}
@@ -217,10 +249,13 @@ public class TableMetaCache {
if (tableMetaTSDB != null) {
// tsdb不需要做,会基于ddl sql自动清理
} else {
for (String name : tableMetaDB.asMap().keySet()) {
if (StringUtils.startsWithIgnoreCase(name, schema + ".")) {
// removeNames.add(name);
tableMetaDB.invalidate(name);
ConcurrentMap<String, TableMeta> map = tableMetaCache.asMap();
if (!map.isEmpty()) {
for (String name : map.keySet()) {
if (StringUtils.startsWithIgnoreCase(name, schema + ".")) {
// removeNames.add(name);
tableMetaCache.invalidate(name);
}
}
}
}
@@ -230,7 +265,7 @@ public class TableMetaCache {
if (tableMetaTSDB != null) {
// tsdb不需要做,会基于ddl sql自动清理
} else {
tableMetaDB.invalidateAll();
tableMetaCache.invalidateAll();
}
}
@@ -252,14 +287,10 @@ public class TableMetaCache {
}
private String getFullName(String schema, String table) {
StringBuilder builder = new StringBuilder();
return builder.append('`')
.append(schema)
.append('`')
StringBuilder builder = new StringBuilder(64);
return builder.append('`').append(schema).append('`')
.append('.')
.append('`')
.append(StringUtils.replace(table,"`","``"))
.append('`')
.append('`').append(StringUtils.replace(table,"`","``")).append('`')
.toString();
}
@@ -286,5 +317,4 @@ public class TableMetaCache {
public void setOnPolarX(boolean isOnPolarX) {
this.isOnPolarX = isOnPolarX;
}
}

1
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java
View File

@@ -300,6 +300,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
@Override
public boolean rollback(EntryPosition position) {
// 每次rollback需要重新构建一次memory data
// FIXME The global MemoryTableMeta may cause OOM when it store too many tables
this.memoryTableMeta = new MemoryTableMeta();
boolean flag = false;
EntryPosition snapshotPosition = buildMemFromSnapshot(position);