diff --git a/README.md b/README.md index 8941c8fb..36dead59 100644 --- a/README.md +++ b/README.md @@ -82,7 +82,7 @@ See the wiki page for : wiki文 1. canal整体交互协议设计上使用了protobuf3.0，理论上可以支持绝大部分的多语言场景，欢迎大家提交多客户端的PR * canal java客户端: https://github.com/alibaba/canal/wiki/ClientExample - * canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp + * canal c#客户端开源项目地址： https://github.com/dotnetcore/CanalSharp * canal go客户端开源项目地址： https://github.com/CanalClient/canal-go 2. canal作为MySQL binlog的增量获取工具，可以将数据投递到MQ系统中，比如Kafka/RocketMQ，可以借助于MQ的多语言能力 * 参考文档: [Canal Kafka/RocketMQ QuickStart](https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart) @@ -117,7 +117,7 @@ See the wiki page for : wiki文

最新更新

  1. canal发布重大版本更新1.1.0，具体releaseNode参考：https://github.com/alibaba/canal/releases/tag/canal-1.1.0
    2. -
  2. canal c#客户端开源项目地址： https://github.com/CanalClient/CanalSharp ，推荐!
    3. +
  3. canal c#客户端开源项目地址： https://github.com/dotnetcore/CanalSharp ，推荐!
  4. canal QQ讨论群已经建立，群号：161559791 ，欢迎加入进行技术讨论。
  5. canal消费端项目开源: Otter(分布式数据库同步系统)，地址：https://github.com/alibaba/otter
    6. diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java index 77e22415..ebcb4ec0 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java @@ -76,7 +76,7 @@ public class JdbcTypeUtil { || "TEXT".equalsIgnoreCase(columnType) || "TINYTEXT".equalsIgnoreCase(columnType); } - public static Object typeConvert(String columnName, String value, int sqlType, String mysqlType) { + public static Object typeConvert(String tableName ,String columnName, String value, int sqlType, String mysqlType) { if (value == null || (value.equals("") && !(isText(mysqlType) || sqlType == Types.CHAR || sqlType == Types.VARCHAR || sqlType == Types.LONGVARCHAR))) { return null; @@ -161,7 +161,7 @@ public class JdbcTypeUtil { } return res; } catch (Exception e) { - logger.error("table: {} column: {}, failed convert type {} to {}", columnName, value, sqlType); + logger.error("table: {} column: {}, failed convert type {} to {}", tableName, columnName, value, sqlType); return value; } } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java index 51e6a1b4..1e24d9ad 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java @@ -77,7 +77,7 @@ public class MessageUtil { } } row.put(column.getName(), - JdbcTypeUtil.typeConvert(column.getName(), + JdbcTypeUtil.typeConvert(dml.getTable(),column.getName(), column.getValue(), column.getSqlType(), column.getMysqlType())); @@ -95,7 +95,7 @@ public class MessageUtil { for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { if (updateSet.contains(column.getName())) { rowOld.put(column.getName(), - JdbcTypeUtil.typeConvert(column.getName(), + JdbcTypeUtil.typeConvert(dml.getTable(),column.getName(), column.getValue(), column.getSqlType(), column.getMysqlType())); @@ -153,16 +153,16 @@ public class MessageUtil { // } List> data = flatMessage.getData(); if (data != null) { - dml.setData(changeRows(data, flatMessage.getSqlType(), flatMessage.getMysqlType())); + dml.setData(changeRows(dml.getTable(), data, flatMessage.getSqlType(), flatMessage.getMysqlType())); } List> old = flatMessage.getOld(); if (old != null) { - dml.setOld(changeRows(old, flatMessage.getSqlType(), flatMessage.getMysqlType())); + dml.setOld(changeRows(dml.getTable(), old, flatMessage.getSqlType(), flatMessage.getMysqlType())); } return dml; } - private static List> changeRows(List> rows, Map sqlTypes, + private static List> changeRows(String table, List> rows, Map sqlTypes, Map mysqlTypes) { List> result = new ArrayList<>(); for (Map row : rows) { @@ -181,7 +181,7 @@ public class MessageUtil { continue; } - Object finalValue = JdbcTypeUtil.typeConvert(columnName, columnValue, sqlType, mysqlType); + Object finalValue = JdbcTypeUtil.typeConvert(table, columnName, columnValue, sqlType, mysqlType); resultRow.put(columnName, finalValue); } result.add(resultRow); diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index 2c3ecfa9..8c9dd469 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java @@ -107,8 +107,9 @@ public class RdbSyncService { int j = i; futures.add(executorThreads[i].submit(() -> { try { - dmlsPartition[j] - .forEach(syncItem -> sync(batchExecutors[j], syncItem.config, syncItem.singleDml)); + dmlsPartition[j].forEach(syncItem -> sync(batchExecutors[j], + syncItem.config, + syncItem.singleDml)); dmlsPartition[j].clear(); batchExecutors[j].commit(); return true; @@ -146,49 +147,49 @@ public class RdbSyncService { sync(dmls, dml -> { if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) { // DDL - columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable()); - return false; + columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable()); + return false; + } else { + // DML + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String groupId = StringUtils.trimToEmpty(dml.getGroupId()); + String database = dml.getDatabase(); + String table = dml.getTable(); + Map configMap; + if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { + configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table); } else { - // DML - String destination = StringUtils.trimToEmpty(dml.getDestination()); - String groupId = StringUtils.trimToEmpty(dml.getGroupId()); - String database = dml.getDatabase(); - String table = dml.getTable(); - Map configMap; - if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { - configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table); - } else { - configMap = mappingConfig.get(destination + "_" + database + "-" + table); - } - - if (configMap == null) { - return false; - } - - if (configMap.values().isEmpty()) { - return false; - } - - for (MappingConfig config : configMap.values()) { - if (config.getConcurrent()) { - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - int hash = pkHash(config.getDbMapping(), singleDml.getData()); - SyncItem syncItem = new SyncItem(config, singleDml); - dmlsPartition[hash].add(syncItem); - }); - } else { - int hash = 0; - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - SyncItem syncItem = new SyncItem(config, singleDml); - dmlsPartition[hash].add(syncItem); - }); - } - } - return true; + configMap = mappingConfig.get(destination + "_" + database + "-" + table); } - }); + + if (configMap == null) { + return false; + } + + if (configMap.values().isEmpty()) { + return false; + } + + for (MappingConfig config : configMap.values()) { + if (config.getConcurrent()) { + List singleDmls = SingleDml.dml2SingleDmls(dml); + singleDmls.forEach(singleDml -> { + int hash = pkHash(config.getDbMapping(), singleDml.getData()); + SyncItem syncItem = new SyncItem(config, singleDml); + dmlsPartition[hash].add(syncItem); + }); + } else { + int hash = 0; + List singleDmls = SingleDml.dml2SingleDmls(dml); + singleDmls.forEach(singleDml -> { + SyncItem syncItem = new SyncItem(config, singleDml); + dmlsPartition[hash].add(syncItem); + }); + } + } + return true; + } + } ); } /** @@ -314,7 +315,7 @@ public class RdbSyncService { for (String srcColumnName : old.keySet()) { List targetColumnNames = new ArrayList<>(); columnsMap.forEach((targetColumn, srcColumn) -> { - if (srcColumnName.toLowerCase().equals(srcColumn.toLowerCase())) { + if (srcColumnName.equalsIgnoreCase(srcColumn)) { targetColumnNames.add(targetColumn); } }); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java index 703de818..70aba4f1 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java @@ -106,8 +106,8 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement // stage 3 SequenceBarrier dmlParserSequenceBarrier = disruptorMsgBuffer.newBarrier(simpleParserStage.getSequence()); - WorkHandler[] workHandlers = new DmlParserStage[parserThreadCount]; - for (int i = 0; i < parserThreadCount; i++) { + WorkHandler[] workHandlers = new DmlParserStage[tc]; + for (int i = 0; i < tc; i++) { workHandlers[i] = new DmlParserStage(); } workerPool = new WorkerPool(disruptorMsgBuffer,