From f07a075dcead5e3124ac66af391a61b6334fb739 Mon Sep 17 00:00:00 2001 From: sunxien <1025488853@qq.com> Date: Wed, 28 Jan 2026 11:52:25 +0800 Subject: [PATCH] =?UTF-8?q?refactor(canal/server):=20=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E7=A8=8B=E5=BA=8F=E4=BB=A3=E7=A0=81=20(#5545)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - canal server logback.xml 配置增加 CanalServerWithEmbedded 单独日志配置，流量非常大的情况下，getMessages/ack等日志量非常大 日志刷盘太快也影响性能，占磁盘空间。用户可以根据需要单独调整日志级别。默认是INFO - TableMetaCache 锁的范围太大，针对这块做了代码逻辑优化。以下场景影响性能： - 当用户关闭tsdb.enable=false, 分库分表数量非常多回查次数很多； - 每张表有四五百列甚至更多，里面的集合容器提前初始化，减少扩容的次数； - DatabaseTableMeta 标记一个内存不安全的地方，以后看情况优化掉 使用MemoryTableMeta在内存区记录表元数据，当分库分表数量非常多，表结构非常复杂非常吃内存，容易OOM --- deployer/src/main/resources/logback.xml | 4 + .../inbound/mysql/dbsync/TableMetaCache.java | 234 ++++++++++-------- .../inbound/mysql/tsdb/DatabaseTableMeta.java | 1 + 3 files changed, 137 insertions(+), 102 deletions(-) diff --git a/deployer/src/main/resources/logback.xml b/deployer/src/main/resources/logback.xml index 1d30442e..9c447cfc 100644 --- a/deployer/src/main/resources/logback.xml +++ b/deployer/src/main/resources/logback.xml @@ -86,6 +86,10 @@ + + + + diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java index f2cd0a64..931676ea 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java @@ -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 tableMetaDB; + private LoadingCache tableMetaCache; public TableMetaCache(MysqlConnection con, TableMetaTSDB tableMetaTSDB){ this.connection = con; this.tableMetaTSDB = tableMetaTSDB; // 如果持久存储的表结构为空，从db里面获取下 if (tableMetaTSDB == null) { - this.tableMetaDB = CacheBuilder.newBuilder().build(new CacheLoader() { - - @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 createLocalCache() { + return CacheBuilder.newBuilder().build(new CacheLoader() { + + @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 fieldMetas = showCreateTable ? parseTableMeta(schema, table, packet) : parseTableMetaByDesc(packet); + return new TableMeta(schema, table, fieldMetas); } public static List parseTableMeta(String schema, String table, ResultSetPacket packet) { @@ -122,86 +139,101 @@ public class TableMetaCache { * 处理desc table的结果 */ public static List parseTableMetaByDesc(ResultSetPacket packet) { - Map nameMaps = new HashMap<>(6, 1f); + List fieldPackets = packet.getFieldDescriptors(); + int size = fieldPackets.size(); + Map 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 result = new ArrayList<>(); + List fieldValues = packet.getFieldValues(); + // int size = packet.getFieldDescriptors().size(); + // int count = packet.getFieldValues().size() / packet.getFieldDescriptors().size(); + int count = fieldValues.size() / size; + List 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 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; } - } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java index f7862692..234de43f 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java @@ -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);