From c75c24808664ede8faadabd082b7a191f3c232fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Wed, 24 Oct 2018 13:45:58 +0800 Subject: [PATCH 1/8] fixed issue #1019 , support canal embedded with Entry --- .../otter/canal/protocol/FlatMessage.java | 16 +++- .../alibaba/otter/canal/protocol/Message.java | 1 - .../embedded/CanalServerWithEmbedded.java | 83 ++++++++++++++----- .../canal/sink/entry/EntryEventSink.java | 10 ++- .../memory/MemoryEventStoreWithBuffer.java | 11 +++ .../otter/canal/store/model/Event.java | 28 ++++++- 6 files changed, 116 insertions(+), 33 deletions(-) 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 dfa4280b..d8821f46 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 @@ -149,11 +149,19 @@ public class FlatMessage implements Serializable { } List flatMessages = new ArrayList<>(); + List entrys = null; + if (message.isRaw()) { + List rawEntries = message.getRawEntries(); + entrys = new ArrayList(rawEntries.size()); + for (ByteString byteString : rawEntries) { + CanalEntry.Entry entry = CanalEntry.Entry.parseFrom(byteString); + entrys.add(entry); + } + } else { + entrys = message.getEntries(); + } - List rawEntries = message.getRawEntries(); - - for (ByteString byteString : rawEntries) { - CanalEntry.Entry entry = CanalEntry.Entry.parseFrom(byteString); + for (CanalEntry.Entry entry : entrys) { if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) { continue; diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index 35adf15d..751e09d0 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -19,7 +19,6 @@ public class Message implements Serializable { private static final long serialVersionUID = 1234034768477580009L; private long id; - @Deprecated private List entries = new ArrayList(); // row data for performance, see: // https://github.com/alibaba/canal/issues/726 diff --git a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java index a4f74c90..b07062ee 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java @@ -1,11 +1,12 @@ package com.alibaba.otter.canal.server.embedded; -import java.util.*; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.ServiceLoader; import java.util.concurrent.TimeUnit; -import com.alibaba.otter.canal.spi.CanalMetricsProvider; -import com.alibaba.otter.canal.spi.CanalMetricsService; -import com.alibaba.otter.canal.spi.NopCanalMetricsService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; @@ -14,6 +15,7 @@ import org.springframework.util.CollectionUtils; import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; import com.alibaba.otter.canal.instance.core.CanalInstance; import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator; +import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.position.LogPosition; @@ -22,7 +24,11 @@ import com.alibaba.otter.canal.protocol.position.PositionRange; import com.alibaba.otter.canal.server.CanalServer; import com.alibaba.otter.canal.server.CanalService; import com.alibaba.otter.canal.server.exception.CanalServerException; +import com.alibaba.otter.canal.spi.CanalMetricsProvider; +import com.alibaba.otter.canal.spi.CanalMetricsService; +import com.alibaba.otter.canal.spi.NopCanalMetricsService; import com.alibaba.otter.canal.store.CanalEventStore; +import com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer; import com.alibaba.otter.canal.store.model.Event; import com.alibaba.otter.canal.store.model.Events; import com.google.common.base.Function; @@ -40,12 +46,12 @@ import com.google.protobuf.ByteString; */ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer, CanalService { - private static final Logger logger = LoggerFactory.getLogger(CanalServerWithEmbedded.class); + private static final Logger logger = LoggerFactory.getLogger(CanalServerWithEmbedded.class); private Map canalInstances; // private Map lastRollbackPostions; private CanalInstanceGenerator canalInstanceGenerator; private int metricsPort; - private CanalMetricsService metrics = NopCanalMetricsService.NOP; + private CanalMetricsService metrics = NopCanalMetricsService.NOP; private static class SingletonHolder { @@ -207,7 +213,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C * b. 如果timeout不为null * 1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回 * 2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少 - * + * * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题) * */ @@ -239,12 +245,23 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C } else { // 记录到流式信息 Long batchId = canalInstance.getMetaManager().addBatch(clientIdentity, events.getPositionRange()); - List entrys = Lists.transform(events.getEvents(), new Function() { + boolean raw = isRaw(canalInstance.getEventStore()); + List entrys = null; + if (raw) { + entrys = Lists.transform(events.getEvents(), new Function() { - public ByteString apply(Event input) { - return input.getRawEntry(); - } - }); + public ByteString apply(Event input) { + return input.getRawEntry(); + } + }); + } else { + entrys = Lists.transform(events.getEvents(), new Function() { + + public CanalEntry.Entry apply(Event input) { + return input.getEntry(); + } + }); + } if (logger.isInfoEnabled()) { logger.info("get successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]", clientIdentity.getClientId(), @@ -255,7 +272,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C } // 直接提交ack ack(clientIdentity, batchId); - return new Message(batchId, true, entrys); + return new Message(batchId, raw, entrys); } } } @@ -283,7 +300,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C * b. 如果timeout不为null * 1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回 * 2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少 - * + * * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题) * */ @@ -311,7 +328,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C } if (CollectionUtils.isEmpty(events.getEvents())) { - // logger.debug("getWithoutAck successfully, clientId:{} batchSize:{} but result + // logger.debug("getWithoutAck successfully, clientId:{} + // batchSize:{} but result // is null", // clientIdentity.getClientId(), // batchSize); @@ -319,12 +337,23 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C } else { // 记录到流式信息 Long batchId = canalInstance.getMetaManager().addBatch(clientIdentity, events.getPositionRange()); - List entrys = Lists.transform(events.getEvents(), new Function() { + boolean raw = isRaw(canalInstance.getEventStore()); + List entrys = null; + if (raw) { + entrys = Lists.transform(events.getEvents(), new Function() { - public ByteString apply(Event input) { - return input.getRawEntry(); - } - }); + public ByteString apply(Event input) { + return input.getRawEntry(); + } + }); + } else { + entrys = Lists.transform(events.getEvents(), new Function() { + + public CanalEntry.Entry apply(Event input) { + return input.getEntry(); + } + }); + } if (logger.isInfoEnabled()) { logger.info("getWithoutAck successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]", clientIdentity.getClientId(), @@ -333,7 +362,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C batchId, events.getPositionRange()); } - return new Message(batchId, true, entrys); + return new Message(batchId, raw, entrys); } } @@ -515,17 +544,25 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C // 发现provider, 进行初始化 if (list.size() > 1) { logger.warn("Found more than one CanalMetricsProvider, use the first one."); - //报告冲突 + // 报告冲突 for (CanalMetricsProvider p : list) { logger.warn("Found CanalMetricsProvider: {}.", p.getClass().getName()); } } - //默认使用第一个 + // 默认使用第一个 CanalMetricsProvider provider = list.get(0); this.metrics = provider.getService(); } } + private boolean isRaw(CanalEventStore eventStore) { + if (eventStore instanceof MemoryEventStoreWithBuffer) { + return ((MemoryEventStoreWithBuffer) eventStore).isRaw(); + } + + return true; + } + // ========= setter ========== public void setCanalInstanceGenerator(CanalInstanceGenerator canalInstanceGenerator) { diff --git a/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java b/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java index 76124ec4..d34450ed 100644 --- a/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java +++ b/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java @@ -20,6 +20,7 @@ import com.alibaba.otter.canal.sink.CanalEventDownStreamHandler; import com.alibaba.otter.canal.sink.CanalEventSink; import com.alibaba.otter.canal.sink.exception.CanalSinkException; import com.alibaba.otter.canal.store.CanalEventStore; +import com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer; import com.alibaba.otter.canal.store.model.Event; /** @@ -42,7 +43,8 @@ public class EntryEventSink extends AbstractCanalEventSink props = entry.getHeader().getPropsList(); if (props != null) { @@ -60,6 +64,16 @@ public class Event implements Serializable { } } } + + if (raw) { + // build raw + this.rawEntry = entry.toByteString(); + this.rawLength = rawEntry.size(); + } else { + this.entry = entry; + // 按照3倍的event length预估 + this.rawLength = entry.getHeader().getEventLength() * 3; + } } public LogIdentity getLogIdentity() { @@ -150,6 +164,14 @@ public class Event implements Serializable { this.rowsCount = rowsCount; } + public CanalEntry.Entry getEntry() { + return entry; + } + + public void setEntry(CanalEntry.Entry entry) { + this.entry = entry; + } + public String toString() { return ToStringBuilder.reflectionToString(this, CanalToStringStyle.DEFAULT_STYLE); } From 575dd59c7893e6a41f8b0da6b9f5a107580d40d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Wed, 24 Oct 2018 16:25:28 +0800 Subject: [PATCH 2/8] fixed mysql set(8 bit) type parse --- .../java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java index c1b96c3c..9fe22820 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java @@ -859,7 +859,7 @@ public final class RowsLogBuffer { // 转化为unsign long switch (len) { case 1: - value = buffer.getInt8(); + value = buffer.getUint8(); break; case 2: value = buffer.getUint16(); From 92708346310f679db608a2f64bdde8b22b7d2468 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Wed, 24 Oct 2018 16:46:26 +0800 Subject: [PATCH 3/8] fixed rds ak/sk config for manager --- .../manager/CanalInstanceWithManager.java | 56 +++++++++++++++++-- .../manager/model/CanalParameter.java | 36 ++++++++++++ .../alibaba/otter/canal/protocol/Message.java | 1 + 3 files changed, 89 insertions(+), 4 deletions(-) diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java index 778a40aa..36774d94 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java @@ -6,7 +6,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; -import com.alibaba.otter.canal.meta.FileMixedMetaManager; +import org.apache.commons.lang.BooleanUtils; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -39,7 +39,16 @@ import com.alibaba.otter.canal.parse.inbound.AbstractEventParser; import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser; import com.alibaba.otter.canal.parse.inbound.mysql.LocalBinlogEventParser; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser; -import com.alibaba.otter.canal.parse.index.*; +import com.alibaba.otter.canal.parse.inbound.mysql.rds.RdsBinlogEventParserProxy; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DefaultTableMetaTSDBFactory; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDBBuilder; +import com.alibaba.otter.canal.parse.index.CanalLogPositionManager; +import com.alibaba.otter.canal.parse.index.FailbackLogPositionManager; +import com.alibaba.otter.canal.parse.index.MemoryLogPositionManager; +import com.alibaba.otter.canal.parse.index.MetaLogPositionManager; +import com.alibaba.otter.canal.parse.index.PeriodMixedLogPositionManager; +import com.alibaba.otter.canal.parse.index.ZooKeeperLogPositionManager; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; import com.alibaba.otter.canal.protocol.position.EntryPosition; import com.alibaba.otter.canal.sink.entry.EntryEventSink; @@ -229,7 +238,18 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { private CanalEventParser doInitEventParser(SourcingType type, List dbAddresses) { CanalEventParser eventParser; if (type.isMysql()) { - MysqlEventParser mysqlEventParser = new MysqlEventParser(); + MysqlEventParser mysqlEventParser = null; + if (StringUtils.isNotEmpty(parameters.getRdsAccesskey()) + && StringUtils.isNotEmpty(parameters.getRdsSecretkey()) + && StringUtils.isNotEmpty(parameters.getRdsInstanceId())) { + + mysqlEventParser = new RdsBinlogEventParserProxy(); + ((RdsBinlogEventParserProxy) mysqlEventParser).setAccesskey(parameters.getRdsAccesskey()); + ((RdsBinlogEventParserProxy) mysqlEventParser).setSecretkey(parameters.getRdsSecretkey()); + ((RdsBinlogEventParserProxy) mysqlEventParser).setInstanceId(parameters.getRdsInstanceId()); + } else { + mysqlEventParser = new MysqlEventParser(); + } mysqlEventParser.setDestination(destination); // 编码参数 mysqlEventParser.setConnectionCharset(Charset.forName(parameters.getConnectionCharset())); @@ -273,6 +293,34 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { mysqlEventParser.setFallbackIntervalInSeconds(parameters.getFallbackIntervalInSeconds()); mysqlEventParser.setProfilingEnabled(false); mysqlEventParser.setFilterTableError(parameters.getFilterTableError()); + mysqlEventParser.setIsGTIDMode(BooleanUtils.toBoolean(parameters.getGtidEnable())); + // tsdb + boolean tsdbEnable = BooleanUtils.toBoolean(parameters.getTsdbEnable()); + if (tsdbEnable) { + mysqlEventParser.setEnableTsdb(tsdbEnable); + mysqlEventParser.setTableMetaTSDBFactory(new DefaultTableMetaTSDBFactory() { + + @Override + public void destory(String destination) { + TableMetaTSDBBuilder.destory(destination); + } + + @Override + public TableMetaTSDB build(String destination, String springXml) { + try { + System.setProperty("canal.instance.tsdb.url", parameters.getTsdbJdbcUrl()); + System.setProperty("canal.instance.tsdb.dbUsername", parameters.getTsdbJdbcUserName()); + System.setProperty("canal.instance.tsdb.dbPassword", parameters.getTsdbJdbcPassword()); + + return TableMetaTSDBBuilder.build(destination, "classpath:spring/tsdb/mysql-tsdb.xml"); + } finally { + System.setProperty("canal.instance.tsdb.url", ""); + System.setProperty("canal.instance.tsdb.dbUsername", ""); + System.setProperty("canal.instance.tsdb.dbPassword", ""); + } + } + }); + } eventParser = mysqlEventParser; } else if (type.isLocalBinlog()) { LocalBinlogEventParser localBinlogEventParser = new LocalBinlogEventParser(); @@ -291,8 +339,8 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { parameters.getDbUsername(), parameters.getDbPassword(), parameters.getDefaultDatabaseName())); - } + eventParser = localBinlogEventParser; } else if (type.isOracle()) { throw new CanalException("unsupport SourcingType for " + type); diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java index 40df1485..ece1196e 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java @@ -97,6 +97,10 @@ public class CanalParameter implements Serializable { private String tsdbJdbcUrl; private String tsdbJdbcUserName; private String tsdbJdbcPassword; + private String rdsAccesskey; + private String rdsSecretkey; + private String rdsInstanceId; + private Boolean gtidEnable = Boolean.FALSE; // 是否开启gtid // ================================== 兼容字段处理 private InetSocketAddress masterAddress; // 主库信息 private String masterUsername; // 帐号 @@ -919,6 +923,38 @@ public class CanalParameter implements Serializable { this.tsdbJdbcPassword = tsdbJdbcPassword; } + public String getRdsAccesskey() { + return rdsAccesskey; + } + + public void setRdsAccesskey(String rdsAccesskey) { + this.rdsAccesskey = rdsAccesskey; + } + + public String getRdsSecretkey() { + return rdsSecretkey; + } + + public void setRdsSecretkey(String rdsSecretkey) { + this.rdsSecretkey = rdsSecretkey; + } + + public String getRdsInstanceId() { + return rdsInstanceId; + } + + public void setRdsInstanceId(String rdsInstanceId) { + this.rdsInstanceId = rdsInstanceId; + } + + public Boolean getGtidEnable() { + return gtidEnable; + } + + public void setGtidEnable(Boolean gtidEnable) { + this.gtidEnable = gtidEnable; + } + public String toString() { return ToStringBuilder.reflectionToString(this, CanalToStringStyle.DEFAULT_STYLE); } diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index 751e09d0..bf79ee45 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -38,6 +38,7 @@ public class Message implements Serializable { } else { this.entries = entries == null ? new ArrayList() : entries; } + this.raw = raw; } public Message(long id){ From b5dac60cbb192ff2f9722b4ff3a2c1e4ac103326 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Wed, 24 Oct 2018 17:00:56 +0800 Subject: [PATCH 4/8] fixed cliet-adapter artifactId --- client-adapter/example/pom.xml | 4 ++-- client-adapter/launcher/pom.xml | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/client-adapter/example/pom.xml b/client-adapter/example/pom.xml index 3c68087a..606534e0 100644 --- a/client-adapter/example/pom.xml +++ b/client-adapter/example/pom.xml @@ -9,7 +9,7 @@ 4.0.0 com.alibaba.otter - example + client-adapter.example jar canal client adapter example module for otter ${project.version} @@ -145,4 +145,4 @@ - \ No newline at end of file + diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index 7b94bfc9..76480a70 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -9,7 +9,7 @@ 4.0.0 com.alibaba.otter - launcher + client-adapter.launcher jar canal client adapter launcher module for otter ${project.version} @@ -134,4 +134,4 @@ - \ No newline at end of file + From 346d091d003a8b604ae2a462bc88bcabc4a57317 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Thu, 25 Oct 2018 13:11:34 +0800 Subject: [PATCH 5/8] fixed issue #1032 , support raw config --- deployer/src/main/resources/canal.properties | 1 + .../resources/spring/default-instance.xml | 1 + .../main/resources/spring/file-instance.xml | 1 + .../main/resources/spring/group-instance.xml | 1 + .../main/resources/spring/memory-instance.xml | 1 + .../manager/CanalInstanceWithManager.java | 1 + .../manager/model/CanalParameter.java | 9 +++++++++ .../otter/canal/kafka/CanalKafkaProducer.java | 19 ++++++++----------- .../otter/canal/server/CanalMQStarter.java | 5 +++++ 9 files changed, 28 insertions(+), 11 deletions(-) diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index ea096de8..2a24b66f 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -20,6 +20,7 @@ canal.instance.memory.buffer.size = 16384 canal.instance.memory.buffer.memunit = 1024 ## meory store gets mode used MEMSIZE or ITEMSIZE canal.instance.memory.batch.mode = MEMSIZE +canal.instance.memory.rawEntry = true ## detecing config canal.instance.detecting.enable = false diff --git a/deployer/src/main/resources/spring/default-instance.xml b/deployer/src/main/resources/spring/default-instance.xml index 95ec6a29..fe9cc7a5 100644 --- a/deployer/src/main/resources/spring/default-instance.xml +++ b/deployer/src/main/resources/spring/default-instance.xml @@ -58,6 +58,7 @@ + diff --git a/deployer/src/main/resources/spring/file-instance.xml b/deployer/src/main/resources/spring/file-instance.xml index df30174a..23d8e4bd 100644 --- a/deployer/src/main/resources/spring/file-instance.xml +++ b/deployer/src/main/resources/spring/file-instance.xml @@ -44,6 +44,7 @@ + diff --git a/deployer/src/main/resources/spring/group-instance.xml b/deployer/src/main/resources/spring/group-instance.xml index 71788f66..e614b86c 100644 --- a/deployer/src/main/resources/spring/group-instance.xml +++ b/deployer/src/main/resources/spring/group-instance.xml @@ -41,6 +41,7 @@ + diff --git a/deployer/src/main/resources/spring/memory-instance.xml b/deployer/src/main/resources/spring/memory-instance.xml index 751e9f4b..f785a46e 100644 --- a/deployer/src/main/resources/spring/memory-instance.xml +++ b/deployer/src/main/resources/spring/memory-instance.xml @@ -41,6 +41,7 @@ + diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java index 36774d94..dff18278 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java @@ -147,6 +147,7 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { memoryEventStore.setBufferMemUnit(parameters.getMemoryStorageBufferMemUnit()); memoryEventStore.setBatchMode(BatchMode.valueOf(parameters.getStorageBatchMode().name())); memoryEventStore.setDdlIsolation(parameters.getDdlIsolation()); + memoryEventStore.setRaw(parameters.getMemoryStorageRawEntry()); eventStore = memoryEventStore; } else if (mode.isFile()) { // 后续版本支持 diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java index ece1196e..ca5e1af7 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java @@ -39,6 +39,7 @@ public class CanalParameter implements Serializable { private BatchMode storageBatchMode = BatchMode.MEMSIZE; // 基于大小返回结果 private Integer memoryStorageBufferSize = 16 * 1024; // 内存存储的buffer大小 private Integer memoryStorageBufferMemUnit = 1024; // 内存存储的buffer内存占用单位，默认为1kb + private Boolean memoryStorageRawEntry = Boolean.TRUE; // 内存存储的对象是否启用raw的ByteString模式 private String fileStorageDirectory; // 文件存储的目录位置 private Integer fileStorageStoreCount; // 每个文件store存储的记录数 private Integer fileStorageRollverCount; // store文件的个数 @@ -955,6 +956,14 @@ public class CanalParameter implements Serializable { this.gtidEnable = gtidEnable; } + public Boolean getMemoryStorageRawEntry() { + return memoryStorageRawEntry; + } + + public void setMemoryStorageRawEntry(Boolean memoryStorageRawEntry) { + this.memoryStorageRawEntry = memoryStorageRawEntry; + } + public String toString() { return ToStringBuilder.reflectionToString(this, CanalToStringStyle.DEFAULT_STYLE); } 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 93f5f4ab..654a4b6b 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 @@ -1,24 +1,21 @@ package com.alibaba.otter.canal.kafka; -import java.util.ArrayList; import java.util.List; import java.util.Properties; -import java.util.concurrent.Future; -import com.alibaba.fastjson.serializer.SerializerFeature; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; -import com.alibaba.otter.canal.spi.CanalMQProducer;; +import com.alibaba.otter.canal.spi.CanalMQProducer; /** * kafka producer 主操作类 @@ -104,8 +101,10 @@ public class CanalKafkaProducer implements CanalMQProducer { for (FlatMessage flatMessage : flatMessages) { if (canalDestination.getPartition() != null) { try { - ProducerRecord record = new ProducerRecord(canalDestination - .getTopic(), canalDestination.getPartition(), null, JSON.toJSONString(flatMessage)); + ProducerRecord record = new ProducerRecord(canalDestination.getTopic(), + canalDestination.getPartition(), + null, + JSON.toJSONString(flatMessage)); producer2.send(record); } catch (Exception e) { logger.error(e.getMessage(), e); @@ -123,8 +122,7 @@ public class CanalKafkaProducer implements CanalMQProducer { FlatMessage flatMessagePart = partitionFlatMessage[i]; if (flatMessagePart != null) { try { - ProducerRecord record = new ProducerRecord( - canalDestination.getTopic(), + ProducerRecord record = new ProducerRecord(canalDestination.getTopic(), i, null, JSON.toJSONString(flatMessagePart)); @@ -138,8 +136,7 @@ public class CanalKafkaProducer implements CanalMQProducer { } } else { try { - ProducerRecord record = new ProducerRecord( - canalDestination.getTopic(), + ProducerRecord record = new ProducerRecord(canalDestination.getTopic(), 0, null, JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); diff --git a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java index b3ed906f..413c42b2 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java @@ -57,6 +57,11 @@ public class CanalMQStarter { System.setProperty("canal.instance.filter.transaction.entry", "true"); } + if (properties.getFlatMessage()) { + // 针对flat message模式,设置为raw避免ByteString->Entry的二次解析 + System.setProperty("canal.instance.memory.rawEntry", "false"); + } + // 对应每个instance启动一个worker线程 List destinations = properties.getCanalDestinations(); From 3b1f6ce4fbb0ab00a82d097093d2b7c8958426f3 Mon Sep 17 00:00:00 2001 From: rewerma Date: Fri, 26 Oct 2018 13:25:40 +0800 Subject: [PATCH 6/8] Update README.md --- client-adapter/README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/client-adapter/README.md b/client-adapter/README.md index 47a5dcc4..642bf50b 100644 --- a/client-adapter/README.md +++ b/client-adapter/README.md @@ -224,7 +224,7 @@ $BIGINT 对应PHOENIX里的BIGINT 8字节 $TINYINT 对应PHOENIX里的TINYINT 1字节 $SMALLINT 对应PHOENIX里的SMALLINT 2字节 $FLOAT 对应PHOENIX里的FLOAT 4字节 -$DOUBLE 对应PHOENIX里的DOUBLE 8字节 +$DOUBLE 对应PHOENIX里的DOUBLE 8字节 $BOOLEAN 对应PHOENIX里的BOOLEAN 1字节 $TIME 对应PHOENIX里的TIME 8字节 $DATE 对应PHOENIX里的DATE 8字节 @@ -248,4 +248,4 @@ create 'MYTEST.PERSON', {NAME=>'CF'} java -jar canal-adapter-launcher.jar ``` #### 验证 -修改mysql mytest.person表的数据, 将会自动同步到HBase的MYTEST.PERSON表下面, 并会打出DML的log \ No newline at end of file +修改mysql mytest.person表的数据, 将会自动同步到HBase的MYTEST.PERSON表下面, 并会打出DML的log From 5cfbc6b91d0e55baba1fe525c8c3f3d4f63dd82a Mon Sep 17 00:00:00 2001 From: rewerma Date: Fri, 26 Oct 2018 13:36:59 +0800 Subject: [PATCH 7/8] Update README.md --- client-adapter/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/client-adapter/README.md b/client-adapter/README.md index 642bf50b..31d6b5a1 100644 --- a/client-adapter/README.md +++ b/client-adapter/README.md @@ -68,7 +68,7 @@ adapter.conf: adapterConfigs: # 适配器内部配置列表 - hbase/mytest_person2.yml # 类型/配置文件名, 这里示例就是对应HBase适配器hbase目录下的mytest_person2.yml文件 ``` -## 2.2 适配器管理REST接口 +## 2.2 同步管理REST接口 #### 2.2.1 查询所有订阅同步的canal destination或MQ topic ``` curl http://127.0.0.1:8081/destinations From fb936f896eb6d2b33c1e3ad6a95c6ddfe4b26ffa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Fri, 26 Oct 2018 14:57:52 +0800 Subject: [PATCH 8/8] fixed client-launcher deploy --- client-adapter/example/pom.xml | 149 -------------- .../example/src/main/assembly/dev.xml | 57 ------ .../example/src/main/assembly/release.xml | 57 ------ .../example/src/main/bin/startup.bat | 25 --- .../example/src/main/bin/startup.sh | 100 ---------- client-adapter/example/src/main/bin/stop.sh | 65 ------ .../otter/canal/client/ClientLauncher.java | 66 ------- .../loader/AbstractCanalAdapterWorker.java | 168 ---------------- .../loader/CanalAdapterKafkaWorker.java | 178 ----------------- .../adapter/loader/CanalAdapterLoader.java | 177 ----------------- .../loader/CanalAdapterRocketMQWorker.java | 124 ------------ .../adapter/loader/CanalAdapterWorker.java | 186 ------------------ .../src/main/resources/canal-client.yml | 18 -- .../example/src/main/resources/logback.xml | 47 ----- client-adapter/hbase/pom.xml | 98 +++++---- client-adapter/launcher/pom.xml | 67 +++++-- .../main/assembly/release.xml} | 0 client-adapter/logger/pom.xml | 58 +++--- client-adapter/pom.xml | 1 - 19 files changed, 139 insertions(+), 1502 deletions(-) delete mode 100644 client-adapter/example/pom.xml delete mode 100644 client-adapter/example/src/main/assembly/dev.xml delete mode 100644 client-adapter/example/src/main/assembly/release.xml delete mode 100644 client-adapter/example/src/main/bin/startup.bat delete mode 100644 client-adapter/example/src/main/bin/startup.sh delete mode 100644 client-adapter/example/src/main/bin/stop.sh delete mode 100644 client-adapter/example/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java delete mode 100644 client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java delete mode 100644 client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java delete mode 100644 client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java delete mode 100644 client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java delete mode 100644 client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java delete mode 100644 client-adapter/example/src/main/resources/canal-client.yml delete mode 100644 client-adapter/example/src/main/resources/logback.xml rename client-adapter/launcher/{assembly.xml => src/main/assembly/release.xml} (100%) diff --git a/client-adapter/example/pom.xml b/client-adapter/example/pom.xml deleted file mode 100644 index bf68d731..00000000 --- a/client-adapter/example/pom.xml +++ /dev/null @@ -1,149 +0,0 @@ - - - - canal.client-adapter - com.alibaba.otter - 1.1.1-SNAPSHOT - - 4.0.0 - com.alibaba.otter - client-adapter.example - jar - canal client adapter example module for otter ${project.version} - - - - com.alibaba.otter - client-adapter.common - ${project.version} - - - com.alibaba.otter - canal.client - ${canal_version} - - - - org.apache.rocketmq - rocketmq-client - 4.3.0 - - - - org.apache.kafka - kafka-clients - 1.1.1 - - - org.yaml - snakeyaml - 1.17 - - - - - com.alibaba.otter - client-adapter.logger - ${project.version} - - - - - - - - maven-jar-plugin - 3.0.2 - - - true - - - **/logback.xml - **/canal-client.yml - - - - - - org.apache.maven.plugins - maven-assembly-plugin - - 2.2.1 - - - assemble - - single - - package - - - - false - false - - - - - - - - dev - - true - - env - !release - - - - - - - maven-assembly-plugin - - - - ${basedir}/src/main/assembly/dev.xml - - canal_client - ${project.build.directory} - - - - - - - - - release - - - env - release - - - - - - - maven-assembly-plugin - - - - ${basedir}/src/main/assembly/release.xml - - - ${project.artifactId}-${project.version} - - ${project.parent.build.directory} - - - - - - - diff --git a/client-adapter/example/src/main/assembly/dev.xml b/client-adapter/example/src/main/assembly/dev.xml deleted file mode 100644 index a3cbc20c..00000000 --- a/client-adapter/example/src/main/assembly/dev.xml +++ /dev/null @@ -1,57 +0,0 @@ - - dist - - dir - - false - - - . - / - - README* - - - - ./src/main/bin - bin - - **/* - - 0755 - - - ./src/main/conf - /conf - - **/* - - - - ./src/main/resources - /conf - - **/* - - - META-INF/** - - - - target - logs - - **/* - - - - - - lib - - junit:junit - - - - diff --git a/client-adapter/example/src/main/assembly/release.xml b/client-adapter/example/src/main/assembly/release.xml deleted file mode 100644 index 9350da6e..00000000 --- a/client-adapter/example/src/main/assembly/release.xml +++ /dev/null @@ -1,57 +0,0 @@ - - dist - - tar.gz - - false - - - . - / - - README* - - - - ./src/main/bin - bin - - **/* - - 0755 - - - ./src/main/conf - /conf - - **/* - - - - ./src/main/resources - /conf - - **/* - - - META-INF/** - - - - target - logs - - **/* - - - - - - lib - - junit:junit - - - - diff --git a/client-adapter/example/src/main/bin/startup.bat b/client-adapter/example/src/main/bin/startup.bat deleted file mode 100644 index 33fd7945..00000000 --- a/client-adapter/example/src/main/bin/startup.bat +++ /dev/null @@ -1,25 +0,0 @@ -@echo off -@if not "%ECHO%" == "" echo %ECHO% -@if "%OS%" == "Windows_NT" setlocal - -set ENV_PATH=.\ -if "%OS%" == "Windows_NT" set ENV_PATH=%~dp0% - -set conf_dir=%ENV_PATH%\..\conf -set canal_conf=%conf_dir%\canal-client.yml -set logback_configurationFile=%conf_dir%\logback.xml - -set CLASSPATH=%conf_dir% -set CLASSPATH=%conf_dir%\..\lib\*;%CLASSPATH% - -set JAVA_MEM_OPTS= -Xms128m -Xmx512m -XX:PermSize=128m -set JAVA_OPTS_EXT= -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dapplication.codeset=UTF-8 -Dfile.encoding=UTF-8 -set JAVA_DEBUG_OPT= -server -Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=9099,server=y,suspend=n -set CANAL_OPTS= -DappName=otter-canal -Dlogback.configurationFile="%logback_configurationFile%" -Dcanal.conf="%canal_conf%" - -set JAVA_OPTS= %JAVA_MEM_OPTS% %JAVA_OPTS_EXT% %JAVA_DEBUG_OPT% %CANAL_OPTS% - -set CMD_STR= java %JAVA_OPTS% -classpath "%CLASSPATH%" java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.client.ClientLauncher -echo start cmd : %CMD_STR% - -java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.client.ClientLauncher \ No newline at end of file diff --git a/client-adapter/example/src/main/bin/startup.sh b/client-adapter/example/src/main/bin/startup.sh deleted file mode 100644 index 0e09ddff..00000000 --- a/client-adapter/example/src/main/bin/startup.sh +++ /dev/null @@ -1,100 +0,0 @@ -#!/bin/bash - -current_path=`pwd` -case "`uname`" in - Linux) - bin_abs_path=$(readlink -f $(dirname $0)) - ;; - *) - bin_abs_path=`cd $(dirname $0); pwd` - ;; -esac -base=${bin_abs_path}/.. -canal_conf=$base/conf/canal-client.yml -logback_configurationFile=$base/conf/logback.xml -export LANG=en_US.UTF-8 -export BASE=$base - -if [ -f $base/bin/canal_client.pid ] ; then - echo "found canal_client.pid , Please run stop.sh first ,then startup.sh" 2>&2 - exit 1 -fi - -## set java path -if [ -z "$JAVA" ] ; then - JAVA=$(which java) -fi - -ALIBABA_JAVA="/usr/alibaba/java/bin/java" -TAOBAO_JAVA="/opt/taobao/java/bin/java" -if [ -z "$JAVA" ]; then - if [ -f $ALIBABA_JAVA ] ; then - JAVA=$ALIBABA_JAVA - elif [ -f $TAOBAO_JAVA ] ; then - JAVA=$TAOBAO_JAVA - else - echo "Cannot find a Java JDK. Please set either set JAVA or put java (>=1.5) in your PATH." 2>&2 - exit 1 - fi -fi - -case "$#" -in -0 ) - ;; -1 ) - var=$* - if [ -f $var ] ; then - canal_conf=$var - else - echo "THE PARAMETER IS NOT CORRECT.PLEASE CHECK AGAIN." - exit - fi;; -2 ) - var=$1 - if [ -f $var ] ; then - canal_conf=$var - else - if [ "$1" = "debug" ]; then - DEBUG_PORT=$2 - DEBUG_SUSPEND="n" - JAVA_DEBUG_OPT="-Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=$DEBUG_PORT,server=y,suspend=$DEBUG_SUSPEND" - fi - fi;; -* ) - echo "THE PARAMETERS MUST BE TWO OR LESS.PLEASE CHECK AGAIN." - exit;; -esac - -str=`file -L $JAVA | grep 64-bit` -if [ -n "$str" ]; then - JAVA_OPTS="-server -Xms2048m -Xmx3072m -Xmn1024m -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -Xss256k -XX:-UseAdaptiveSizePolicy -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+UseCMSCompactAtFullCollection -XX:+UseFastAccessorMethods -XX:+UseCMSInitiatingOccupancyOnly -XX:+HeapDumpOnOutOfMemoryError" -else - JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m " -fi - -JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dfile.encoding=UTF-8" -CANAL_OPTS="-DappName=otter-canal-client -Dlogback.configurationFile=$logback_configurationFile -Dcanal.conf=$canal_conf" - -if [ -e $canal_conf -a -e $logback_configurationFile ] -then - - for i in $base/lib/*; - do CLASSPATH=$i:"$CLASSPATH"; - done - CLASSPATH="$base/conf:$CLASSPATH"; - - echo "cd to $bin_abs_path for workaround relative path" - cd $bin_abs_path - - echo LOG CONFIGURATION : $logback_configurationFile - echo canal conf : $canal_conf - echo CLASSPATH :$CLASSPATH - $JAVA $JAVA_OPTS $JAVA_DEBUG_OPT $CANAL_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.client.ClientLauncher 1>>$base/logs/canal_client.log 2>&1 & - echo $! > $base/bin/canal_client.pid - - echo "cd to $current_path for continue" - cd $current_path -else - echo "canal client conf("$canal_conf") OR log configration file($logback_configurationFile) is not exist,please create then first!" -fi diff --git a/client-adapter/example/src/main/bin/stop.sh b/client-adapter/example/src/main/bin/stop.sh deleted file mode 100644 index c96f2807..00000000 --- a/client-adapter/example/src/main/bin/stop.sh +++ /dev/null @@ -1,65 +0,0 @@ -#!/bin/bash - -cygwin=false; -linux=false; -case "`uname`" in - CYGWIN*) - cygwin=true - ;; - Linux*) - linux=true - ;; -esac - -get_pid() { - STR=$1 - PID=$2 - if $cygwin; then - JAVA_CMD="$JAVA_HOME\bin\java" - JAVA_CMD=`cygpath --path --unix $JAVA_CMD` - JAVA_PID=`ps |grep $JAVA_CMD |awk '{print $1}'` - else - if $linux; then - if [ ! -z "$PID" ]; then - JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep "$PID"|grep -v grep|awk '{print $2}'` - else - JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep -v grep|awk '{print $2}'` - fi - else - if [ ! -z "$PID" ]; then - JAVA_PID=`ps aux |grep "$STR"|grep "$PID"|grep -v grep|awk '{print $2}'` - else - JAVA_PID=`ps aux |grep "$STR"|grep -v grep|awk '{print $2}'` - fi - fi - fi - echo $JAVA_PID; -} - -base=`dirname $0`/.. -pidfile=$base/bin/canal_client.pid -if [ ! -f "$pidfile" ];then - echo "canal client is not running. exists" - exit -fi - -pid=`cat $pidfile` -if [ "$pid" == "" ] ; then - pid=`get_pid "appName=otter-canal-client"` -fi - -echo -e "`hostname`: stopping canal $pid ... " -kill $pid - -LOOPS=0 -while (true); -do - gpid=`get_pid "appName=otter-canal-client" "$pid"` - if [ "$gpid" == "" ] ; then - echo "Oook! cost:$LOOPS" - `rm $pidfile` - break; - fi - let LOOPS=LOOPS+1 - sleep 1 -done \ No newline at end of file diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java deleted file mode 100644 index a705a526..00000000 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java +++ /dev/null @@ -1,66 +0,0 @@ -package com.alibaba.otter.canal.client; - -import java.io.FileInputStream; - -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.yaml.snakeyaml.Yaml; - -import com.alibaba.otter.canal.client.adapter.loader.CanalAdapterLoader; -import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; - -public class ClientLauncher { - - private static final String CLASSPATH_URL_PREFIX = "classpath:"; - private static final Logger logger = LoggerFactory.getLogger(ClientLauncher.class); - - public static void main(String[] args) { - try { - logger.info("## set default uncaught exception handler"); - setGlobalUncaughtExceptionHandler(); - - logger.info("## load canal client configurations"); - String conf = System.getProperty("client.conf", "classpath:canal-client.yml"); - CanalClientConfig canalClientConfig; - if (conf.startsWith(CLASSPATH_URL_PREFIX)) { - conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); - canalClientConfig = new Yaml().loadAs(ClientLauncher.class.getClassLoader().getResourceAsStream(conf), - CanalClientConfig.class); - } else { - canalClientConfig = new Yaml().loadAs(new FileInputStream(conf), CanalClientConfig.class); - } - logger.info("## start the canal client adapters."); - final CanalAdapterLoader adapterLoader = new CanalAdapterLoader(canalClientConfig); - adapterLoader.init(); - logger.info("## the canal client adapters are running now ......"); - Runtime.getRuntime().addShutdownHook(new Thread() { - - public void run() { - try { - logger.info("## stop the canal client adapters"); - adapterLoader.destroy(); - } catch (Throwable e) { - logger.warn("## something goes wrong when stopping canal client adapters:", e); - } finally { - logger.info("## canal client adapters are down."); - } - } - - }); - } catch (Throwable e) { - logger.error("## something goes wrong when starting up the canal client adapters:", e); - System.exit(0); - } - } - - private static void setGlobalUncaughtExceptionHandler() { - Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { - - @Override - public void uncaughtException(Thread t, Throwable e) { - logger.error("UnCaughtException", e); - } - }); - } -} diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java deleted file mode 100644 index bbfd462d..00000000 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java +++ /dev/null @@ -1,168 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.loader; - -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.Callable; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Future; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.alibaba.otter.canal.client.adapter.OuterAdapter; -import com.alibaba.otter.canal.client.adapter.support.Dml; -import com.alibaba.otter.canal.client.adapter.support.MessageUtil; -import com.alibaba.otter.canal.protocol.FlatMessage; -import com.alibaba.otter.canal.protocol.Message; - -/** - * 适配器工作线程抽象类 - * - * @author machengyuan 2018-8-19 下午11:30:49 - * @version 1.0.0 - */ -public abstract class AbstractCanalAdapterWorker { - - protected final Logger logger = LoggerFactory.getLogger(this.getClass()); - - protected String canalDestination; // canal实例 - protected List> canalOuterAdapters; // 外部适配器 - protected ExecutorService groupInnerExecutorService; // 组内工作线程池 - protected volatile boolean running = false; // 是否运行中 - protected Thread thread = null; - protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() { - - @Override - public void uncaughtException(Thread t, Throwable e) { - logger.error("parse events has an error", e); - } - }; - - protected void writeOut(final Message message) { - List> futures = new ArrayList<>(); - // 组间适配器并行运行 - for (List outerAdapters : canalOuterAdapters) { - final List adapters = outerAdapters; - futures.add(groupInnerExecutorService.submit(new Callable() { - - @Override - public Boolean call() { - try { - // 组内适配器穿行运行，尽量不要配置组内适配器 - for (final OuterAdapter c : adapters) { - long begin = System.currentTimeMillis(); - MessageUtil.parse4Dml(canalDestination, message, new MessageUtil.Consumer() { - - @Override - public void accept(Dml dml) { - c.sync(dml); - } - }); - - if (logger.isDebugEnabled()) { - logger.debug("{} elapsed time: {}", - c.getClass().getName(), - (System.currentTimeMillis() - begin)); - } - } - return true; - } catch (Exception e) { - return false; - } - } - })); - - // 等待所有适配器写入完成 - // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成 - for (Future f : futures) { - try { - if (!f.get()) { - logger.error("Outer adapter write failed"); - } - } catch (InterruptedException | ExecutionException e) { - // ignore - } - } - } - } - - protected void writeOut(final FlatMessage flatMessage) { - List> futures = new ArrayList<>(); - // 组间适配器并行运行 - for (List outerAdapters : canalOuterAdapters) { - final List adapters = outerAdapters; - futures.add(groupInnerExecutorService.submit(new Callable() { - - @Override - public Boolean call() { - try { - // 组内适配器穿行运行，尽量不要配置组内适配器 - for (OuterAdapter c : adapters) { - long begin = System.currentTimeMillis(); - Dml dml = MessageUtil.flatMessage2Dml(canalDestination, flatMessage); - c.sync(dml); - if (logger.isDebugEnabled()) { - logger.debug("{} elapsed time: {}", - c.getClass().getName(), - (System.currentTimeMillis() - begin)); - } - } - return true; - } catch (Exception e) { - return false; - } - } - })); - - // 等待所有适配器写入完成 - // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成 - for (Future f : futures) { - try { - if (!f.get()) { - logger.error("Outer adapter write failed"); - } - } catch (InterruptedException | ExecutionException e) { - // ignore - } - } - } - } - - protected void writeOut(Message message, String topic) { - if (logger.isDebugEnabled()) { - logger.debug("topic: {} batchId: {} batchSize: {} ", topic, message.getId(), message.getEntries().size()); - } - long begin = System.currentTimeMillis(); - writeOut(message); - long now = System.currentTimeMillis(); - if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) { - logger.error("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin); - } - if (logger.isDebugEnabled()) { - logger.debug("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin); - } - } - - protected void stopOutAdapters() { - if (thread != null) { - try { - thread.join(); - } catch (InterruptedException e) { - // ignore - } - } - groupInnerExecutorService.shutdown(); - logger.info("topic connectors' worker thread dead!"); - for (List outerAdapters : canalOuterAdapters) { - for (OuterAdapter adapter : outerAdapters) { - adapter.destroy(); - } - } - logger.info("topic all connectors destroyed!"); - } - - public abstract void start(); - - public abstract void stop(); -} diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java deleted file mode 100644 index b2d3e583..00000000 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java +++ /dev/null @@ -1,178 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.loader; - -import java.util.List; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.TimeUnit; - -import org.apache.kafka.clients.consumer.CommitFailedException; -import org.apache.kafka.common.errors.WakeupException; - -import com.alibaba.otter.canal.client.adapter.OuterAdapter; -import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector; -import com.alibaba.otter.canal.client.kafka.KafkaCanalConnectors; -import com.alibaba.otter.canal.protocol.FlatMessage; -import com.alibaba.otter.canal.protocol.Message; - -/** - * kafka对应的client适配器工作线程 - * - * @author machengyuan 2018-8-19 下午11:30:49 - * @version 1.0.0 - */ -public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { - - private KafkaCanalConnector connector; - - private String topic; - - private boolean flatMessage; - - public CanalAdapterKafkaWorker(String bootstrapServers, String topic, String groupId, - List> canalOuterAdapters, boolean flatMessage){ - this.canalOuterAdapters = canalOuterAdapters; - this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); - this.topic = topic; - this.canalDestination = topic; - this.flatMessage = flatMessage; - connector = KafkaCanalConnectors.newKafkaConnector(bootstrapServers, topic, null, groupId, flatMessage); - // connector.setSessionTimeout(1L, TimeUnit.MINUTES); - - // super.initSwitcher(topic); - } - - @Override - public void start() { - if (!running) { - thread = new Thread(new Runnable() { - - @Override - public void run() { - process(); - } - }); - thread.setUncaughtExceptionHandler(handler); - running = true; - thread.start(); - } - } - - @Override - public void stop() { - try { - if (!running) { - return; - } - - connector.stopRunning(); - running = false; - - // if (switcher != null && !switcher.state()) { - // switcher.set(true); - // } - - if (thread != null) { - try { - thread.join(); - } catch (InterruptedException e) { - // ignore - } - } - groupInnerExecutorService.shutdown(); - logger.info("topic {} connectors' worker thread dead!", this.topic); - for (List outerAdapters : canalOuterAdapters) { - for (OuterAdapter adapter : outerAdapters) { - adapter.destroy(); - } - } - logger.info("topic {} all connectors destroyed!", this.topic); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - private void process() { - while (!running) - ; - ExecutorService executor = Executors.newSingleThreadExecutor(); - // final AtomicBoolean executing = new AtomicBoolean(true); - while (running) { - try { - logger.info("=============> Start to connect topic: {} <=============", this.topic); - connector.connect(); - logger.info("=============> Start to subscribe topic: {} <=============", this.topic); - connector.subscribe(); - logger.info("=============> Subscribe topic: {} succeed <=============", this.topic); - while (running) { - try { - // switcher.get(); //等待开关开启 - - List messages; - if (!flatMessage) { - messages = connector.getWithoutAck(); - } else { - messages = connector.getFlatMessageWithoutAck(100L, TimeUnit.MILLISECONDS); - } - if (messages != null) { - for (final Object message : messages) { - if (message instanceof FlatMessage) { - writeOut((FlatMessage) message); - } else { - writeOut((Message) message); - } - // executing.set(true); - // if (message != null) { - // executor.submit(new Runnable() { - // - // @Override - // public void run() { - // try { - // if (message instanceof FlatMessage) { - // writeOut((FlatMessage) message); - // } else { - // writeOut((Message) message); - // } - // } catch (Exception e) { - // logger.error(e.getMessage(), e); - // } finally { - // executing.compareAndSet(true, false); - // } - // } - // }); - // - // // 间隔一段时间ack一次, 防止因超时未响应切换到另外台客户端 - // long currentTS = System.currentTimeMillis(); - // while (executing.get()) { - // // 大于10秒未消费完ack一次keep alive - // if (System.currentTimeMillis() - currentTS > 10000) { - // connector.ack(); - // currentTS = System.currentTimeMillis(); - // } - // } - // } - } - } - connector.ack(); - } catch (CommitFailedException e) { - logger.warn(e.getMessage()); - } catch (Exception e) { - logger.error(e.getMessage(), e); - TimeUnit.SECONDS.sleep(1L); - } - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - executor.shutdown(); - - try { - connector.unsubscribe(); - } catch (WakeupException e) { - // No-op. Continue process - } - connector.disconnect(); - logger.info("=============> Disconnect topic: {} <=============", this.topic); - } -} diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java deleted file mode 100644 index 51b12a0a..00000000 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java +++ /dev/null @@ -1,177 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.loader; - -import java.net.InetSocketAddress; -import java.net.SocketAddress; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; - -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.alibaba.otter.canal.client.adapter.OuterAdapter; -import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; -import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; -import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; - -/** - * MQ外部适配器的加载器 - * - * @version 1.0.0 - */ -public class CanalAdapterLoader { - - private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class); - - private CanalClientConfig canalClientConfig; - - private Map canalWorkers = new HashMap<>(); - - private Map canalMQWorker = new HashMap<>(); - - private ExtensionLoader loader; - - public CanalAdapterLoader(CanalClientConfig canalClientConfig){ - this.canalClientConfig = canalClientConfig; - } - - /** - * 初始化canal-client、 canal-client-rocketmq的适配器 - */ - public void init() { - // canal instances 和 mq topics 配置不能同时为空 - if (canalClientConfig.getCanalInstances() == null && canalClientConfig.getMqTopics() == null) { - throw new RuntimeException("Blank config property: canalInstances or canalMQTopics"); - } - - loader = ExtensionLoader.getExtensionLoader(OuterAdapter.class, "" /* - * TODO - * canalClientConfig - * . - * getClassloaderPolicy - * ( - * ) - */); - - String canalServerHost = this.canalClientConfig.getCanalServerHost(); - SocketAddress sa = null; - if (canalServerHost != null) { - String[] ipPort = canalServerHost.split(":"); - sa = new InetSocketAddress(ipPort[0], Integer.parseInt(ipPort[1])); - } - String zkHosts = this.canalClientConfig.getZookeeperHosts(); - - // 初始化canal-client的适配器 - if (canalClientConfig.getCanalInstances() != null) { - for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) { - List> canalOuterAdapterGroups = new ArrayList<>(); - - for (CanalClientConfig.AdapterGroup connectorGroup : instance.getAdapterGroups()) { - List canalOutConnectors = new ArrayList<>(); - for (OuterAdapterConfig c : connectorGroup.getOutAdapters()) { - loadConnector(c, canalOutConnectors); - } - canalOuterAdapterGroups.add(canalOutConnectors); - } - CanalAdapterWorker worker; - if (zkHosts != null) { - worker = new CanalAdapterWorker(instance.getInstance(), zkHosts, canalOuterAdapterGroups); - } else { - worker = new CanalAdapterWorker(instance.getInstance(), sa, canalOuterAdapterGroups); - } - canalWorkers.put(instance.getInstance(), worker); - worker.start(); - logger.info("Start adapter for canal instance: {} succeed", instance.getInstance()); - } - } - - // 初始化canal-client-mq的适配器 - if (canalClientConfig.getMqTopics() != null) { - for (CanalClientConfig.MQTopic topic : canalClientConfig.getMqTopics()) { - for (CanalClientConfig.Group group : topic.getGroups()) { - List> canalOuterAdapterGroups = new ArrayList<>(); - - List canalOuterAdapters = new ArrayList<>(); - - for (OuterAdapterConfig config : group.getOutAdapters()) { - loadConnector(config, canalOuterAdapters); - } - canalOuterAdapterGroups.add(canalOuterAdapters); - if (StringUtils.isBlank(topic.getMqMode()) || "rocketmq".equalsIgnoreCase(topic.getMqMode())) { - CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker(canalClientConfig.getBootstrapServers(), - topic.getTopic(), - group.getGroupId(), - canalOuterAdapterGroups); - canalMQWorker.put(topic.getTopic() + "-rocketmq-" + group.getGroupId(), rocketMQWorker); - rocketMQWorker.start(); - } else if ("kafka".equalsIgnoreCase(topic.getMqMode())) { - CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(canalClientConfig.getBootstrapServers(), - topic.getTopic(), - group.getGroupId(), - canalOuterAdapterGroups, - canalClientConfig.getFlatMessage()); - canalMQWorker.put(topic.getTopic() + "-kafka-" + group.getGroupId(), canalKafkaWorker); - canalKafkaWorker.start(); - } - logger.info("Start adapter for canal-client rocketmq topic: {} succeed", topic.getTopic() + "-" - + group.getGroupId()); - - } - } - } - } - - private void loadConnector(OuterAdapterConfig config, List canalOutConnectors) { - try { - OuterAdapter adapter = loader.getExtension(config.getName()); - ClassLoader cl = Thread.currentThread().getContextClassLoader(); - // 替换ClassLoader - Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader()); - adapter.init(config); - Thread.currentThread().setContextClassLoader(cl); - canalOutConnectors.add(adapter); - logger.info("Load canal adapter: {} succeed", config.getName()); - } catch (Exception e) { - logger.error("Load canal adapter: {} failed", config.getName(), e); - } - } - - /** - * 销毁所有适配器 为防止canal实例太多造成销毁阻塞, 并行销毁 - */ - public void destroy() { - if (canalWorkers.size() > 0) { - ExecutorService stopExecutorService = Executors.newFixedThreadPool(canalWorkers.size()); - for (CanalAdapterWorker v : canalWorkers.values()) { - final CanalAdapterWorker caw = v; - stopExecutorService.submit(new Runnable() { - - @Override - public void run() { - caw.stop(); - } - }); - } - stopExecutorService.shutdown(); - } - if (canalMQWorker.size() > 0) { - ExecutorService stopMQWokerService = Executors.newFixedThreadPool(canalMQWorker.size()); - for (AbstractCanalAdapterWorker tmp : canalMQWorker.values()) { - final AbstractCanalAdapterWorker worker = tmp; - stopMQWokerService.submit(new Runnable() { - - @Override - public void run() { - worker.stop(); - } - }); - } - stopMQWokerService.shutdown(); - } - logger.info("All canal adapters destroyed"); - } -} diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java deleted file mode 100644 index 9c126297..00000000 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java +++ /dev/null @@ -1,124 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.loader; - -import java.util.List; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.TimeUnit; - -import org.apache.kafka.clients.consumer.CommitFailedException; -import org.apache.kafka.common.errors.WakeupException; - -import com.alibaba.otter.canal.client.adapter.OuterAdapter; -import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; -import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnectorProvider; -import com.alibaba.otter.canal.protocol.Message; - -/** - * kafka对应的client适配器工作线程 - * - * @author machengyuan 2018-8-19 下午11:30:49 - * @version 1.0.0 - */ -public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { - - private RocketMQCanalConnector connector; - - private String topic; - - public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId, - List> canalOuterAdapters){ - logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId); - this.canalOuterAdapters = canalOuterAdapters; - this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); - this.topic = topic; - this.canalDestination = topic; - connector = RocketMQCanalConnectorProvider.newRocketMQConnector(nameServers, topic, groupId); - } - - @Override - public void start() { - if (!running) { - thread = new Thread(new Runnable() { - - @Override - public void run() { - process(); - } - }); - thread.setUncaughtExceptionHandler(handler); - running = true; - thread.start(); - } - } - - @Override - public void stop() { - try { - if (!running) { - return; - } - connector.stopRunning(); - running = false; - logger.info("Stop topic {} out adapters begin", this.topic); - stopOutAdapters(); - logger.info("Stop topic {} out adapters end", this.topic); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - private void process() { - while (!running) - ; - ExecutorService executor = Executors.newSingleThreadExecutor(); - while (running) { - try { - logger.info("=============> Start to connect topic: {} <=============", this.topic); - connector.connect(); - logger.info("=============> Start to subscribe topic: {}<=============", this.topic); - connector.subscribe(); - logger.info("=============> Subscribe topic: {} succeed<=============", this.topic); - while (running) { - try { - // switcher.get(); //等待开关开启 - - final Message message = connector.getWithoutAck(1); - if (message != null) { - executor.submit(new Runnable() { - - @Override - public void run() { - try { - writeOut(message, topic); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - connector.ack(message.getId()); - } - }); - } else { - logger.debug("Message is null"); - } - } catch (CommitFailedException e) { - logger.warn(e.getMessage()); - } catch (Exception e) { - logger.error(e.getMessage(), e); - TimeUnit.SECONDS.sleep(1L); - } - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - executor.shutdown(); - - try { - connector.unsubscribe(); - } catch (WakeupException e) { - // No-op. Continue process - } - connector.stopRunning(); - logger.info("=============> Disconnect topic: {} <=============", this.topic); - } -} diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java deleted file mode 100644 index e72ed2d5..00000000 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java +++ /dev/null @@ -1,186 +0,0 @@ -package com.alibaba.otter.canal.client.adapter.loader; - -import java.net.SocketAddress; -import java.util.List; -import java.util.concurrent.Executors; - -import com.alibaba.otter.canal.client.CanalConnector; -import com.alibaba.otter.canal.client.CanalConnectors; -import com.alibaba.otter.canal.client.adapter.OuterAdapter; -import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; -import com.alibaba.otter.canal.protocol.Message; - -/** - * 原生canal-server对应的client适配器工作线程 - * - * @author machengyuan 2018-8-19 下午11:30:49 - * @version 1.0.0 - */ -public class CanalAdapterWorker extends AbstractCanalAdapterWorker { - - private static final int BATCH_SIZE = 50; - private static final int SO_TIMEOUT = 0; - - private CanalConnector connector; - - /** - * 单台client适配器worker的构造方法 - * - * @param canalDestination canal实例名 - * @param address canal-server地址 - * @param canalOuterAdapters 外部适配器组 - */ - public CanalAdapterWorker(String canalDestination, SocketAddress address, - List> canalOuterAdapters){ - this.canalOuterAdapters = canalOuterAdapters; - this.canalDestination = canalDestination; - groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); - connector = CanalConnectors.newSingleConnector(address, canalDestination, "", ""); - } - - /** - * HA模式下client适配器worker的构造方法 - * - * @param canalDestination canal实例名 - * @param zookeeperHosts zookeeper地址 - * @param canalOuterAdapters 外部适配器组 - */ - public CanalAdapterWorker(String canalDestination, String zookeeperHosts, - List> canalOuterAdapters){ - this.canalOuterAdapters = canalOuterAdapters; - this.canalDestination = canalDestination; - groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); - connector = CanalConnectors.newClusterConnector(zookeeperHosts, canalDestination, "", ""); - ((ClusterCanalConnector) connector).setSoTimeout(SO_TIMEOUT); - - // super.initSwitcher(canalDestination); - } - - @Override - public void start() { - if (!running) { - thread = new Thread(new Runnable() { - - @Override - public void run() { - process(); - } - }); - thread.setUncaughtExceptionHandler(handler); - thread.start(); - running = true; - } - } - - @Override - public void stop() { - try { - if (!running) { - return; - } - - // if (switcher != null && !switcher.state()) { - // switcher.set(true); - // } - - connector.stopRunning(); - running = false; - - logger.info("destination {} is waiting for adapters' worker thread die!", canalDestination); - if (thread != null) { - try { - thread.join(); - } catch (InterruptedException e) { - // ignore - } - } - groupInnerExecutorService.shutdown(); - logger.info("destination {} adapters' worker thread dead!", canalDestination); - for (List outerAdapters : canalOuterAdapters) { - for (OuterAdapter adapter : outerAdapters) { - adapter.destroy(); - } - } - logger.info("destination {} all adapters destroyed!", canalDestination); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - private void process() { - while (!running) - ; // waiting until running == true - while (running) { - try { - // if (switcher != null) { - // switcher.get(); - // } - logger.info("=============> Start to connect destination: {} <=============", this.canalDestination); - connector.connect(); - logger.info("=============> Start to subscribe destination: {} <=============", this.canalDestination); - connector.subscribe(); - logger.info("=============> Subscribe destination: {} succeed <=============", this.canalDestination); - while (running) { - // try { - // if (switcher != null) { - // switcher.get(); - // } - // } catch (TimeoutException e) { - // break; - // } - - // server配置canal.instance.network.soTimeout(默认: 30s) - // 范围内未与server交互，server将关闭本次socket连接 - Message message = connector.getWithoutAck(BATCH_SIZE); // 获取指定数量的数据 - long batchId = message.getId(); - try { - int size = message.getEntries().size(); - - if (batchId == -1 || size == 0) { - try { - Thread.sleep(1000); - } catch (InterruptedException e) { - // ignore - } - } else { - if (logger.isDebugEnabled()) { - logger.debug("destination: {} batchId: {} batchSize: {} ", - this.canalDestination, - batchId, - size); - } - long begin = System.currentTimeMillis(); - writeOut(message); - long now = System.currentTimeMillis(); - if (logger.isDebugEnabled()) { - logger.debug("destination: {} batchId: {} elapsed time: {} ms", - this.canalDestination, - batchId, - now - begin); - } - } - connector.ack(batchId); // 提交确认 - } catch (Exception e) { - connector.rollback(batchId); // 处理失败, 回滚数据 - throw e; - } - } - - } catch (Exception e) { - logger.error("process error!", e); - } finally { - connector.disconnect(); - logger.info("=============> Disconnect destination: {} <=============", this.canalDestination); - } - - if (running) { // is reconnect - try { - Thread.sleep(1000); - } catch (InterruptedException e) { - // ignore - } - } - - } - } -} diff --git a/client-adapter/example/src/main/resources/canal-client.yml b/client-adapter/example/src/main/resources/canal-client.yml deleted file mode 100644 index de24e756..00000000 --- a/client-adapter/example/src/main/resources/canal-client.yml +++ /dev/null @@ -1,18 +0,0 @@ -canalServerHost: 127.0.0.1:11111 -#zookeeperHosts: slave1:2181 -#bootstrapServers: slave1:6667,slave2:6667 #or rocketmq nameservers:host1:9876;host2:9876 -flatMessage: false - -canalInstances: -- instance: example - adapterGroups: - - outAdapters: - - name: logger - -#mqTopics: -#- mqMode: rocketmq -# topic: example -# groups: -# - groupId: example -# outAdapters: -# - name: logger diff --git a/client-adapter/example/src/main/resources/logback.xml b/client-adapter/example/src/main/resources/logback.xml deleted file mode 100644 index e74c84f0..00000000 --- a/client-adapter/example/src/main/resources/logback.xml +++ /dev/null @@ -1,47 +0,0 @@ - - - - - %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n - - - - - - - destination - canal_client - - - - ../logs/${destination}.log - - - ../logs/%d{yyyy-MM-dd}/${destination}-%d{yyyy-MM-dd}-%i.log.gz - - - 512MB - - 60 - - - - %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n - - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/client-adapter/hbase/pom.xml b/client-adapter/hbase/pom.xml index a48d7401..f48cdd10 100644 --- a/client-adapter/hbase/pom.xml +++ b/client-adapter/hbase/pom.xml @@ -43,48 +43,60 @@ - - - - org.apache.maven.plugins - maven-assembly-plugin - 2.4 - - - jar-with-dependencies - - - - - make-assembly - package - - single - - - - - - maven-antrun-plugin - - - package - - run - - - - - - - - - - - - - - - + + + release + + + env + release + + + + + + + org.apache.maven.plugins + maven-assembly-plugin + 2.4 + + + jar-with-dependencies + + + + + make-assembly + package + + single + + + + + + maven-antrun-plugin + + + package + + run + + + + + + + + + + + + + + + + + diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index cb8aee72..f10dfaac 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -104,6 +104,18 @@ org.springframework.boot spring-boot-maven-plugin 2.0.1.RELEASE + + + + com.alibaba.otter + client-adapter.hbase + + + com.alibaba.otter + client-adapter.logger + + + @@ -150,24 +162,43 @@ - - maven-assembly-plugin - - false - - assembly.xml - - - - - make-assembly - package - - single - - - - + + + + release + + + env + release + + + + + + + maven-assembly-plugin + + false + + ${basedir}/src/main/assembly/release.xml + + ${project.artifactId}-${project.version} + ${project.basedir}/../../target + + + + make-assembly + package + + single + + + + + + + + diff --git a/client-adapter/launcher/assembly.xml b/client-adapter/launcher/src/main/assembly/release.xml similarity index 100% rename from client-adapter/launcher/assembly.xml rename to client-adapter/launcher/src/main/assembly/release.xml diff --git a/client-adapter/logger/pom.xml b/client-adapter/logger/pom.xml index 8f2b4edb..e9a146aa 100644 --- a/client-adapter/logger/pom.xml +++ b/client-adapter/logger/pom.xml @@ -21,27 +21,39 @@ - - - - org.apache.maven.plugins - maven-assembly-plugin - 2.4 - - - jar-with-dependencies - - - - - make-assembly - package - - single - - - - - - + + + release + + + env + release + + + + + + + org.apache.maven.plugins + maven-assembly-plugin + 2.4 + + + jar-with-dependencies + + + + + make-assembly + package + + single + + + + + + + + diff --git a/client-adapter/pom.xml b/client-adapter/pom.xml index 7ca65de1..aafe6395 100644 --- a/client-adapter/pom.xml +++ b/client-adapter/pom.xml @@ -24,7 +24,6 @@ logger hbase launcher - example