From 0cda90cdeae57c4d30468ffa581e105c784695da Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 20 Sep 2018 16:17:38 +0800 Subject: [PATCH 1/9] =?UTF-8?q?=E9=BB=98=E8=AE=A4=E5=B0=86=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=8F=91=E7=8E=B0kafka=E7=9A=84partition=200=EF=BC=8C?= =?UTF-8?q?=E9=81=BF=E5=85=8D=E5=A4=9A=E5=88=86=E5=8C=BA=E7=9A=84=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E4=B8=80=E8=87=B4=E6=80=A7=E9=97=AE=E9=A2=98=20?= =?UTF-8?q?=E5=90=8E=E7=BB=AD=E5=B0=86=E6=94=AF=E6=8C=81=E9=80=9A=E8=BF=87?= =?UTF-8?q?id=20hash=E5=88=86=E5=8F=91=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/kafka/CanalKafkaProducer.java | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 4942a37b..e1ea5e47 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -75,7 +75,7 @@ public class CanalKafkaProducer { if (topic.getPartition() != null) { record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); } else { - record = new ProducerRecord(topic.getTopic(), message); + record = new ProducerRecord(topic.getTopic(), 0, null, message); } producer.send(record); @@ -84,8 +84,18 @@ public class CanalKafkaProducer { List flatMessages = FlatMessage.messageConverter(message); if (flatMessages != null) { for (FlatMessage flatMessage : flatMessages) { - ProducerRecord record = new ProducerRecord(topic.getTopic(), - JSON.toJSONString(flatMessage)); + ProducerRecord record; + if (topic.getPartition() != null) { + record = new ProducerRecord(topic.getTopic(), + topic.getPartition(), + null, + JSON.toJSONString(flatMessage)); + } else { + record = new ProducerRecord(topic.getTopic(), + 0, + null, + JSON.toJSONString(flatMessage)); + } producer2.send(record); } } From 74890050d0c323877eb9c1e3dec128e989eb6ee8 Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 20 Sep 2018 16:48:14 +0800 Subject: [PATCH 2/9] =?UTF-8?q?kafka=E7=94=9F=E4=BA=A7=E7=AB=AF=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=E6=8C=89pk=20hash=E5=88=B0=E5=AF=B9=E5=BA=94partition?= =?UTF-8?q?=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deployer/src/main/resources/kafka.yml | 6 +- .../otter/canal/protocol/FlatMessage.java | 31 ++++++---- .../otter/canal/kafka/CanalKafkaProducer.java | 55 ++++++++++++----- .../otter/canal/kafka/CanalKafkaStarter.java | 5 +- .../otter/canal/kafka/KafkaProperties.java | 60 +++++-------------- 5 files changed, 79 insertions(+), 78 deletions(-) diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml index 735f723d..b86b3034 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -12,7 +12,9 @@ flatMessage: true canalDestinations: - canalDestination: example - topic: example - partition: + topic: expample + #partitionsNum: 3 + #partitionHash: + # mytest.person: id diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java index d7150b79..51bd320a 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java @@ -1,14 +1,8 @@ package com.alibaba.otter.canal.protocol; import java.io.Serializable; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; +import java.util.*; -import com.google.common.collect.Table; import com.google.protobuf.ByteString; /** @@ -31,7 +25,7 @@ public class FlatMessage implements Serializable { private List> data; private List> old; - public FlatMessage() { + public FlatMessage(){ } public FlatMessage(long id){ @@ -126,6 +120,12 @@ public class FlatMessage implements Serializable { this.old = old; } + /** + * 将Message转换为FlatMessage + * + * @param message 原生message + * @return FlatMessage列表 + */ public static List messageConverter(Message message) { try { if (message == null) { @@ -231,11 +231,22 @@ public class FlatMessage implements Serializable { } } + /** + * 将FlatMessage按指定的字段值hash拆分 + * + * @param flatMessage flatMessage + * @param partitionsNum 分区数量 + * @param pkHashConfig hash映射 + * @return 拆分后的flatMessage数组 + */ public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, - Table pkHashConfig) { + Map pkHashConfig) { + if (partitionsNum == null) { + partitionsNum = 1; + } FlatMessage[] partitionMessages = new FlatMessage[partitionsNum]; - String pk = pkHashConfig.get(flatMessage.getDatabase(), flatMessage.getTable()); + String pk = pkHashConfig.get(flatMessage.getDatabase() + "." + flatMessage.getTable()); if (pk == null || flatMessage.getIsDdl()) { partitionMessages[0] = flatMessage; } else { diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index e1ea5e47..fbeeeb87 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -67,15 +67,18 @@ public class CanalKafkaProducer { } } - public void send(KafkaProperties.Topic topic, Message message, Callback callback) { + public void send(KafkaProperties.CanalDestination canalDestination, Message message, Callback callback) { try { // producer.beginTransaction(); if (!kafkaProperties.getFlatMessage()) { ProducerRecord record; - if (topic.getPartition() != null) { - record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); + if (canalDestination.getPartition() != null) { + record = new ProducerRecord(canalDestination.getTopic(), + canalDestination.getPartition(), + null, + message); } else { - record = new ProducerRecord(topic.getTopic(), 0, null, message); + record = new ProducerRecord(canalDestination.getTopic(), 0, null, message); } producer.send(record); @@ -84,19 +87,39 @@ public class CanalKafkaProducer { List flatMessages = FlatMessage.messageConverter(message); if (flatMessages != null) { for (FlatMessage flatMessage : flatMessages) { - ProducerRecord record; - if (topic.getPartition() != null) { - record = new ProducerRecord(topic.getTopic(), - topic.getPartition(), - null, - JSON.toJSONString(flatMessage)); + if (canalDestination.getPartition() != null) { + ProducerRecord record = new ProducerRecord(canalDestination + .getTopic(), canalDestination.getPartition(), null, JSON.toJSONString(flatMessage)); + producer2.send(record); } else { - record = new ProducerRecord(topic.getTopic(), - 0, - null, - JSON.toJSONString(flatMessage)); + if (canalDestination.getPartitionHash() != null + && !canalDestination.getPartitionHash().isEmpty()) { + FlatMessage[] partitionFlatMessage = FlatMessage.messagePartition(flatMessage, + canalDestination.getPartitionsNum(), + canalDestination.getPartitionHash()); + int length = partitionFlatMessage.length; + for (int i = 0; i < length; i++) { + FlatMessage flatMessage1 = partitionFlatMessage[i]; + if (flatMessage1 == null) { + continue; + } + ProducerRecord record = new ProducerRecord( + canalDestination.getTopic(), + i, + null, + JSON.toJSONString(partitionFlatMessage[i])); + producer2.send(record); + } + } else { + ProducerRecord record = new ProducerRecord( + canalDestination.getTopic(), + 0, + null, + JSON.toJSONString(flatMessage)); + producer2.send(record); + } } - producer2.send(record); + } } } @@ -104,7 +127,7 @@ public class CanalKafkaProducer { // producer.commitTransaction(); callback.commit(); if (logger.isDebugEnabled()) { - logger.debug("send message to kafka topic: {}", topic.getTopic()); + logger.debug("send message to kafka topic: {}", canalDestination.getTopic()); } } catch (Exception e) { logger.error(e.getMessage(), e); diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java index 5b73bf3b..abe7bf09 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java @@ -132,10 +132,7 @@ public class CanalKafkaStarter implements CanalServerStarter { try { int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size(); if (batchId != -1 && size != 0) { - Topic topic = new Topic(); - topic.setTopic(destination.getTopic()); - topic.setPartition(destination.getPartition()); - canalKafkaProducer.send(topic, message, new CanalKafkaProducer.Callback() { + canalKafkaProducer.send(destination, message, new CanalKafkaProducer.Callback() { @Override public void commit() { diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java b/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java index 82707ba4..5e4145fe 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java @@ -1,9 +1,8 @@ package com.alibaba.otter.canal.kafka; import java.util.ArrayList; -import java.util.HashSet; import java.util.List; -import java.util.Set; +import java.util.Map; /** * kafka 配置项 @@ -27,10 +26,11 @@ public class KafkaProperties { public static class CanalDestination { - private String canalDestination; - private String topic; - private Integer partition; - private Set topics = new HashSet(); + private String canalDestination; + private String topic; + private Integer partition; + private Integer partitionsNum; + private Map partitionHash; public String getCanalDestination() { return canalDestination; @@ -56,52 +56,20 @@ public class KafkaProperties { this.partition = partition; } - public Set getTopics() { - return topics; + public Integer getPartitionsNum() { + return partitionsNum; } - public void setTopics(Set topics) { - this.topics = topics; - } - } - - public static class Topic { - - private String topic; - private Integer partition; - - public String getTopic() { - return topic; + public void setPartitionsNum(Integer partitionsNum) { + this.partitionsNum = partitionsNum; } - public void setTopic(String topic) { - this.topic = topic; + public Map getPartitionHash() { + return partitionHash; } - public Integer getPartition() { - return partition; - } - - public void setPartition(Integer partition) { - this.partition = partition; - } - - @Override - public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - - Topic topic1 = (Topic) o; - - if (topic != null ? !topic.equals(topic1.topic) : topic1.topic != null) return false; - return partition != null ? partition.equals(topic1.partition) : topic1.partition == null; - } - - @Override - public int hashCode() { - int result = topic != null ? topic.hashCode() : 0; - result = 31 * result + (partition != null ? partition.hashCode() : 0); - return result; + public void setPartitionHash(Map partitionHash) { + this.partitionHash = partitionHash; } } From e566160f1ea24c3db15b45f51b1b6ad2ce226b42 Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 20 Sep 2018 17:24:32 +0800 Subject: [PATCH 3/9] =?UTF-8?q?=E6=95=B4=E7=90=86=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deployer/src/main/resources/kafka.yml | 10 ++++++---- .../otter/canal/kafka/CanalKafkaProducer.java | 17 ++++++++--------- 2 files changed, 14 insertions(+), 13 deletions(-) diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml index b86b3034..76e2bade 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -12,9 +12,11 @@ flatMessage: true canalDestinations: - canalDestination: example - topic: expample - #partitionsNum: 3 - #partitionHash: - # mytest.person: id + topic: example +# #对应topic分区数量 +# partitionsNum: 3 +# partitionHash: +# #库名.表名: 唯一主键 +# mytest.person: id diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index fbeeeb87..5708105c 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -99,16 +99,15 @@ public class CanalKafkaProducer { canalDestination.getPartitionHash()); int length = partitionFlatMessage.length; for (int i = 0; i < length; i++) { - FlatMessage flatMessage1 = partitionFlatMessage[i]; - if (flatMessage1 == null) { - continue; + FlatMessage flatMessagePart = partitionFlatMessage[i]; + if (flatMessagePart != null) { + ProducerRecord record = new ProducerRecord( + canalDestination.getTopic(), + i, + null, + JSON.toJSONString(flatMessagePart)); + producer2.send(record); } - ProducerRecord record = new ProducerRecord( - canalDestination.getTopic(), - i, - null, - JSON.toJSONString(partitionFlatMessage[i])); - producer2.send(record); } } else { ProducerRecord record = new ProducerRecord( From 4f2b4cea53e9dfc526f6600fbf1c0a3bf955dabd Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 20 Sep 2018 17:25:32 +0800 Subject: [PATCH 4/9] =?UTF-8?q?=E6=95=B4=E7=90=86=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deployer/src/main/resources/kafka.yml | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml index 76e2bade..aa23c1ee 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -11,12 +11,12 @@ canalGetTimeout: 100 flatMessage: true canalDestinations: - - canalDestination: example - topic: example -# #对应topic分区数量 -# partitionsNum: 3 -# partitionHash: -# #库名.表名: 唯一主键 -# mytest.person: id +- canalDestination: example + topic: exp3 +# #对应topic分区数量 +# partitionsNum: 3 +# partitionHash: +# #库名.表名: 唯一主键 +# mytest.person: id From 19b7f730b640471150796f487f002378dca0cc0b Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 20 Sep 2018 18:06:47 +0800 Subject: [PATCH 5/9] =?UTF-8?q?=E6=95=B4=E7=90=86=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java | 1 - 1 file changed, 1 deletion(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java index abe7bf09..f2df3547 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java @@ -12,7 +12,6 @@ import org.slf4j.LoggerFactory; import org.yaml.snakeyaml.Yaml; import com.alibaba.otter.canal.kafka.KafkaProperties.CanalDestination; -import com.alibaba.otter.canal.kafka.KafkaProperties.Topic; import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.server.CanalServerStarter; From fd80f953e97eaf741ace487f950c8d2492ea60a8 Mon Sep 17 00:00:00 2001 From: agapple Date: Fri, 21 Sep 2018 15:30:35 +0800 Subject: [PATCH 6/9] Update README.md --- README.md | 44 +++++++++++++++++++++++++++++--------------- 1 file changed, 29 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 86ef8755..2d7bc589 100644 --- a/README.md +++ b/README.md @@ -5,6 +5,7 @@
  1. canal QQ讨论群已经建立，群号：161559791 ，欢迎加入进行技术讨论。
  2. canal消费端项目开源: Otter(分布式数据库同步系统)，地址：https://github.com/alibaba/otter
    3. +
  3. canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp ，推荐!
  4. Canal已在阿里云推出商业化版本 数据传输服务DTS， 开通即用，免去部署维护的昂贵使用成本。DTS针对阿里云RDS、DRDS等产品进行了适配，解决了Binlog日志回收，主备切换、VPC网络切换等场景下的订阅高可用问题。同时，针对RDS进行了针对性的性能优化。出于稳定性、性能及成本的考虑，强烈推荐阿里云用户使用DTS产品。DTS产品使用文档
    5. DTS支持阿里云RDS&DRDS的Binlog日志实时订阅，现推出首月免费体验，限时限量，立即体验>>>
@@ -52,18 +53,38 @@ See the wiki page for : wiki文

wiki文档列表

+

多语言业务使用

+ +1. canal整体交互协议设计上使用了protobuf3.0，理论上可以支持绝大部分的多语言场景，欢迎大家提交多客户端的PR + * canal java客户端: https://github.com/alibaba/canal/wiki/ClientExample + * canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp + * canal go客户端，开发进行中 +2. canal作为MySQL binlog的增量获取工具，可以将数据投递到MQ系统中，比如Kafka/RocketMQ，可以借助于MQ的多语言能力 +

相关资料

* ADC阿里技术嘉年华分享ppt (放在google docs上，可能需要翻墙): ppt下载 @@ -90,10 +111,3 @@ See the wiki page for : wiki文
  • 新浪微博： agapple0002
  • 报告issue：issues
    • - -
    -【招聘】阿里巴巴中间件团队招聘JAVA高级工程师
-岗位主要为技术型内容(非业务部门)，阿里中间件整个体系对于未来想在技术上有所沉淀的同学还是非常有帮助的
-工作地点：杭州、北京均可. ps. 阿里待遇向来都是不错的，有意者可以QQ、微博私聊. 
-具体招聘内容：https://job.alibaba.com/zhaopin/position_detail.htm?positionId=32666
-
    From 9535e82da6a3ed760919747122f25b8ee1ae334e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Fri, 21 Sep 2018 15:40:18 +0800 Subject: [PATCH 7/9] fixed readme --- README.md | 17 +++++++++++++++-- .../resources/spring/tsdb/sql/create_table.sql | 8 ++++---- 2 files changed, 19 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 2d7bc589..95331c55 100644 --- a/README.md +++ b/README.md @@ -3,9 +3,11 @@

    最新更新

      +
    1. canal发布重大版本更新1.1.0，具体releaseNode参考：https://github.com/alibaba/canal/releases/tag/canal-1.1.0
      2. +
    2. canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp ，推荐!
    3. canal QQ讨论群已经建立，群号：161559791 ，欢迎加入进行技术讨论。
    4. canal消费端项目开源: Otter(分布式数据库同步系统)，地址：https://github.com/alibaba/otter
      5. -
    5. canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp ，推荐!
      6. +
    6. Canal已在阿里云推出商业化版本 数据传输服务DTS， 开通即用，免去部署维护的昂贵使用成本。DTS针对阿里云RDS、DRDS等产品进行了适配，解决了Binlog日志回收，主备切换、VPC网络切换等场景下的订阅高可用问题。同时，针对RDS进行了针对性的性能优化。出于稳定性、性能及成本的考虑，强烈推荐阿里云用户使用DTS产品。DTS产品使用文档
      7. DTS支持阿里云RDS&DRDS的Binlog日志实时订阅，现推出首月免费体验，限时限量，立即体验>>>
    @@ -47,6 +49,16 @@ DTS支持阿里云RDS&DRDS的Binlog日志实时订阅，现推出首月免费体
  • canal解析binary log对象(原始为byte流)
    • +

    重要版本更新说明

    + +canal 1.1.x系列，参考release文档：版本发布信息： + +1. 整体性能测试&优化,提升了150%. #726 参考: 【[Performance](https://github.com/alibaba/canal/wiki/Performance)】 +2. 原生支持prometheus监控 #765 【[Prometheus QuickStart](https://github.com/alibaba/canal/wiki/Prometheus-QuickStart)】 +3. 原生支持kafka消息投递 #695 【[Canal Kafka QuickStart](https://github.com/alibaba/canal/wiki/Canal-Kafka-QuickStart)】 +4. 原生支持aliyun rds的binlog订阅 (解决自动主备切换/oss binlog离线解析) 参考: 【[Aliyun RDS QuickStart](https://github.com/alibaba/canal/wiki/aliyun-RDS-QuickStart)】 +5. 原生支持docker镜像 #801 参考: 【[Docker QuickStart](https://github.com/alibaba/canal/wiki/Docker-QuickStart)】 +

    相关文档

    See the wiki page for : wiki文档 @@ -77,7 +89,7 @@ See the wiki page for : wiki文
  • FAQ
    • -

    多语言业务使用

    +

    多语言业务

    1. canal整体交互协议设计上使用了protobuf3.0，理论上可以支持绝大部分的多语言场景，欢迎大家提交多客户端的PR * canal java客户端: https://github.com/alibaba/canal/wiki/ClientExample @@ -111,3 +123,4 @@ See the wiki page for : wiki文
  • 新浪微博： agapple0002
  • 报告issue：issues
    • + diff --git a/deployer/src/main/resources/spring/tsdb/sql/create_table.sql b/deployer/src/main/resources/spring/tsdb/sql/create_table.sql index e5d6dbf4..322bc2b6 100644 --- a/deployer/src/main/resources/spring/tsdb/sql/create_table.sql +++ b/deployer/src/main/resources/spring/tsdb/sql/create_table.sql @@ -26,10 +26,10 @@ CREATE TABLE IF NOT EXISTS `meta_history` ( `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id', `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳', `use_schema` varchar(1024) DEFAULT NULL COMMENT '执行sql时对应的schema', - `schema` varchar(1024) DEFAULT NULL COMMENT '对应的schema', - `table` varchar(1024) DEFAULT NULL COMMENT '对应的table', - `sql` longtext DEFAULT NULL COMMENT '执行的sql', - `type` varchar(256) DEFAULT NULL COMMENT 'sql类型', + `sql_schema` varchar(1024) DEFAULT NULL COMMENT '对应的schema', + `sql_table` varchar(1024) DEFAULT NULL COMMENT '对应的table', + `sql_text` longtext DEFAULT NULL COMMENT '执行的sql', + `sql_type` varchar(256) DEFAULT NULL COMMENT 'sql类型', `extra` text DEFAULT NULL COMMENT '额外的扩展信息', PRIMARY KEY (`id`), UNIQUE KEY binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`), From 9d48d23787c8e24689c7a3c8b431c6109e0a5d4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Fri, 21 Sep 2018 15:44:03 +0800 Subject: [PATCH 8/9] fixed readme --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 95331c55..9a87de8f 100644 --- a/README.md +++ b/README.md @@ -51,7 +51,7 @@ DTS支持阿里云RDS&DRDS的Binlog日志实时订阅，现推出首月免费体

    重要版本更新说明

    -canal 1.1.x系列，参考release文档：版本发布信息： +canal 1.1.x系列，参考release文档：版本发布信息 1. 整体性能测试&优化,提升了150%. #726 参考: 【[Performance](https://github.com/alibaba/canal/wiki/Performance)】 2. 原生支持prometheus监控 #765 【[Prometheus QuickStart](https://github.com/alibaba/canal/wiki/Prometheus-QuickStart)】 From 00ec70d425e5ebbd0dbcb7211c7354325af99be6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Fri, 21 Sep 2018 15:45:52 +0800 Subject: [PATCH 9/9] fixed readme --- README.md | 23 +++++++++++------------ 1 file changed, 11 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 9a87de8f..c28fc0ea 100644 --- a/README.md +++ b/README.md @@ -1,17 +1,6 @@
    - -

    最新更新

    -
      -
    1. canal发布重大版本更新1.1.0，具体releaseNode参考：https://github.com/alibaba/canal/releases/tag/canal-1.1.0
      2. -
    2. canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp ，推荐!
      3. -
    3. canal QQ讨论群已经建立，群号：161559791 ，欢迎加入进行技术讨论。
      4. -
    4. canal消费端项目开源: Otter(分布式数据库同步系统)，地址：https://github.com/alibaba/otter
      5. - -
    5. Canal已在阿里云推出商业化版本 数据传输服务DTS， 开通即用，免去部署维护的昂贵使用成本。DTS针对阿里云RDS、DRDS等产品进行了适配，解决了Binlog日志回收，主备切换、VPC网络切换等场景下的订阅高可用问题。同时，针对RDS进行了针对性的性能优化。出于稳定性、性能及成本的考虑，强烈推荐阿里云用户使用DTS产品。DTS产品使用文档
      6. -DTS支持阿里云RDS&DRDS的Binlog日志实时订阅，现推出首月免费体验，限时限量，立即体验>>> -
    - +

    背景

    早期，阿里巴巴B2B公司因为存在杭州和美国双机房部署，存在跨机房同步的业务需求。不过早期的数据库同步业务，主要是基于trigger的方式获取增量变更，不过从2010年开始，阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&消费的业务，从此开启了一段新纪元。

    ps. 目前内部版本已经支持mysql和oracle部分版本的日志解析，当前的canal开源版本支持5.7及以下的版本(阿里内部mysql 5.7.13, 5.6.10, mysql 5.5.18和5.1.40/48)

    @@ -124,3 +113,13 @@ See the wiki page for : wiki文
  • 报告issue：issues
    • +

    最新更新

    +
      +
    1. canal发布重大版本更新1.1.0，具体releaseNode参考：https://github.com/alibaba/canal/releases/tag/canal-1.1.0
      2. +
    2. canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp ，推荐!
      3. +
    3. canal QQ讨论群已经建立，群号：161559791 ，欢迎加入进行技术讨论。
      4. +
    4. canal消费端项目开源: Otter(分布式数据库同步系统)，地址：https://github.com/alibaba/otter
      5. + +
    5. Canal已在阿里云推出商业化版本 数据传输服务DTS， 开通即用，免去部署维护的昂贵使用成本。DTS针对阿里云RDS、DRDS等产品进行了适配，解决了Binlog日志回收，主备切换、VPC网络切换等场景下的订阅高可用问题。同时，针对RDS进行了针对性的性能优化。出于稳定性、性能及成本的考虑，强烈推荐阿里云用户使用DTS产品。DTS产品使用文档
      6. +DTS支持阿里云RDS&DRDS的Binlog日志实时订阅，现推出首月免费体验，限时限量，立即体验>>> +
    \ No newline at end of file