From 2707c12570de69b9f00ba4ebabfa97d725f77630 Mon Sep 17 00:00:00 2001 From: AlexLEWIS Date: Mon, 11 Mar 2019 14:47:14 +0800 Subject: [PATCH 1/4] CanalSharp has transferred to .NET Core Community CanalSharp has transferred to .NET Core Community(NCC) and became a member project of NCC. --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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. From 078673c6a9215520e1d2a0b098e1df040720abf0 Mon Sep 17 00:00:00 2001 From: kangzhidong Date: Wed, 13 Mar 2019 16:55:38 +0800 Subject: [PATCH 2/4] fix the error log does not contain table name --- .../canal/client/adapter/support/JdbcTypeUtil.java | 4 ++-- .../canal/client/adapter/support/MessageUtil.java | 12 ++++++------ 2 files changed, 8 insertions(+), 8 deletions(-) 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); From df98fec945295514df37137332e3ec72ad64e20b Mon Sep 17 00:00:00 2001 From: lizhyocx Date: Fri, 15 Mar 2019 17:11:14 +0800 Subject: [PATCH 3/4] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E5=8D=95=E6=A0=B8?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1=E5=99=A8=E6=97=A0=E6=B3=95=E8=AE=A2=E9=98=85?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9A=84bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 解决单核服务器c中anal.instance.parser.parallel = true时无法订阅数据的bug --- .../canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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, From 95b26c4d1d44ec377436ac0b383237c3863e82c0 Mon Sep 17 00:00:00 2001 From: agapple Date: Sun, 17 Mar 2019 10:35:16 +0800 Subject: [PATCH 4/4] fixed #1433, optimzier code --- .../adapter/rdb/service/RdbSyncService.java | 89 ++++++++++--------- 1 file changed, 45 insertions(+), 44 deletions(-) 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); } });