From 6ee6b3c3980f422921dce524fdf806286572589f Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 25 Oct 2018 16:43:33 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=8C=E6=AD=A5=E5=88=86=E5=B8=83=E5=BC=8F?= =?UTF-8?q?=E5=BC=80=E5=85=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client-adapter/common/pom.xml | 5 - .../client/adapter/hbase/HbaseAdapter.java | 7 +- client-adapter/launcher/pom.xml | 5 + .../adapter/launcher/common/SyncSwitch.java | 130 ++++++++++++++++-- .../launcher/config/AdapterCanalConfig.java | 2 +- .../launcher/config/CuratorClient.java | 36 +++++ .../launcher/loader/CanalAdapterLoader.java | 8 +- .../launcher/loader/CanalAdapterWorker.java | 2 - .../adapter/launcher/rest/CommonRest.java | 5 +- .../src/main/resources/application.yml | 5 +- 10 files changed, 178 insertions(+), 27 deletions(-) create mode 100644 client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/CuratorClient.java diff --git a/client-adapter/common/pom.xml b/client-adapter/common/pom.xml index d61d8d11..eb3a21a3 100644 --- a/client-adapter/common/pom.xml +++ b/client-adapter/common/pom.xml @@ -18,11 +18,6 @@ canal.protocol ${canal_version} - - org.slf4j - slf4j-api - 1.7.12 - joda-time joda-time diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java index 50eb271c..ababeb6e 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java @@ -8,6 +8,7 @@ import java.util.Map; import javax.sql.DataSource; +import org.apache.commons.lang.StringUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.TableName; @@ -52,7 +53,8 @@ public class HbaseAdapter implements OuterAdapter { hbaseMapping = MappingConfigLoader.load(); mappingConfigCache = new HashMap<>(); for (MappingConfig mappingConfig : hbaseMapping.values()) { - mappingConfigCache.put(mappingConfig.getHbaseOrm().getDatabase() + "-" + mappingConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getHbaseOrm().getDestination()) + + "." + mappingConfig.getHbaseOrm().getDatabase() + "." + mappingConfig.getHbaseOrm().getTable(), mappingConfig); } @@ -77,9 +79,10 @@ public class HbaseAdapter implements OuterAdapter { if (dml == null) { return; } + String destination = StringUtils.trimToEmpty(dml.getDestination()); String database = dml.getDatabase(); String table = dml.getTable(); - MappingConfig config = mappingConfigCache.get(database + "-" + table); + MappingConfig config = mappingConfigCache.get(destination + "." + database + "." + table); hbaseSyncService.sync(config, dml); } diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index 7d54cd58..0f39411e 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -50,6 +50,11 @@ spring-boot-configuration-processor true + + org.apache.curator + curator-recipes + 2.10.0 + org.apache.rocketmq diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java index 2a2f1220..e9e647fd 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java @@ -1,5 +1,6 @@ package com.alibaba.otter.canal.adapter.launcher.common; +import java.nio.charset.StandardCharsets; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; @@ -8,42 +9,114 @@ import java.util.concurrent.TimeoutException; import javax.annotation.PostConstruct; import javax.annotation.Resource; -import org.apache.commons.lang.StringUtils; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.recipes.cache.NodeCache; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.data.Stat; import org.springframework.stereotype.Component; import com.alibaba.otter.canal.adapter.launcher.config.AdapterCanalConfig; +import com.alibaba.otter.canal.adapter.launcher.config.CuratorClient; import com.alibaba.otter.canal.common.utils.BooleanMutex; @Component public class SyncSwitch { - private static final Map LOCAL_LOCK = new ConcurrentHashMap<>(); + private static final String SYN_SWITCH_ZKNODE = "/sync-switch/"; - private Mode mode = Mode.LOCAL; + private static final Map LOCAL_LOCK = new ConcurrentHashMap<>(); + + private static final Map DISTRIBUTED_LOCK = new ConcurrentHashMap<>(); + + private static Mode mode = Mode.LOCAL; @Resource private AdapterCanalConfig adapterCanalConfig; + @Resource + private CuratorClient curatorClient; @PostConstruct public void init() { - if (StringUtils.isEmpty(adapterCanalConfig.getZookeeperHosts())) { + CuratorFramework curator = curatorClient.getCurator(); + if (curator != null) { + mode = Mode.DISTRIBUTED; + DISTRIBUTED_LOCK.clear(); + for (String destination : adapterCanalConfig.DESTINATIONS) { + // 对应每个destination注册锁 + BooleanMutex mutex = new BooleanMutex(true); + initMutex(curator, destination, mutex); + DISTRIBUTED_LOCK.put(destination, mutex); + startListen(destination, mutex); + } + } else { mode = Mode.LOCAL; LOCAL_LOCK.clear(); - for (String destination : AdapterCanalConfig.DESTINATIONS) { + for (String destination : adapterCanalConfig.DESTINATIONS) { // 对应每个destination注册锁 LOCAL_LOCK.put(destination, new BooleanMutex(true)); } - } else { - mode = Mode.DISTRIBUTED; + } + } + + private synchronized void startListen(String destination, BooleanMutex mutex) { + try { + String path = SYN_SWITCH_ZKNODE + destination; + CuratorFramework curator = curatorClient.getCurator(); + final NodeCache nodeCache = new NodeCache(curator, path); + nodeCache.start(); + nodeCache.getListenable().addListener(() -> initMutex(curator, destination, mutex)); + } catch (Exception e) { + throw new RuntimeException(e.getMessage()); + } + } + + private synchronized void initMutex(CuratorFramework curator, String destination, BooleanMutex mutex) { + try { + String path = SYN_SWITCH_ZKNODE + destination; + Stat stat = curator.checkExists().forPath(path); + if (stat == null) { + if (!mutex.state()) { + mutex.set(true); + } + } else { + String data = new String(curator.getData().forPath(path), StandardCharsets.UTF_8); + if ("on".equals(data)) { + if (!mutex.state()) { + mutex.set(true); + } + } else { + if (mutex.state()) { + mutex.set(false); + } + } + } + } catch (Exception e) { + throw new RuntimeException(e.getMessage()); } } public synchronized void off(String destination) { + if (mode == Mode.LOCAL) { BooleanMutex mutex = LOCAL_LOCK.get(destination); if (mutex != null && mutex.state()) { mutex.set(false); } + } else { + try { + String path = SYN_SWITCH_ZKNODE + destination; + try { + curatorClient.getCurator() + .create() + .creatingParentContainersIfNeeded() + .withMode(CreateMode.PERSISTENT) + .forPath(path, "off".getBytes(StandardCharsets.UTF_8)); + } catch (Exception e) { + curatorClient.getCurator().setData().forPath(path, "off".getBytes(StandardCharsets.UTF_8)); + } + } catch (Exception e) { + throw new RuntimeException(e); + } } } @@ -53,14 +126,35 @@ public class SyncSwitch { if (mutex != null && !mutex.state()) { mutex.set(true); } + } else { + try { + String path = SYN_SWITCH_ZKNODE + destination; + try { + curatorClient.getCurator() + .create() + .creatingParentContainersIfNeeded() + .withMode(CreateMode.PERSISTENT) + .forPath(path, "on".getBytes(StandardCharsets.UTF_8)); + } catch (Exception e) { + curatorClient.getCurator().setData().forPath(path, "on".getBytes(StandardCharsets.UTF_8)); + } + } catch (Exception e) { + throw new RuntimeException(e); + } } } public synchronized void release(String destination) { if (mode == Mode.LOCAL) { BooleanMutex mutex = LOCAL_LOCK.get(destination); - if (mutex != null && mutex.state()) { - mutex.set(false); + if (mutex != null && !mutex.state()) { + mutex.set(true); + } + } + if (mode == Mode.DISTRIBUTED) { + BooleanMutex mutex = DISTRIBUTED_LOCK.get(destination); + if (mutex != null && !mutex.state()) { + mutex.set(true); } } } @@ -73,8 +167,14 @@ public class SyncSwitch { } else { return null; } + } else { + BooleanMutex mutex = DISTRIBUTED_LOCK.get(destination); + if (mutex != null) { + return mutex.state(); + } else { + return null; + } } - return null; } public void get(String destination) throws InterruptedException { @@ -83,6 +183,11 @@ public class SyncSwitch { if (mutex != null) { mutex.get(); } + } else { + BooleanMutex mutex = DISTRIBUTED_LOCK.get(destination); + if (mutex != null) { + mutex.get(); + } } } @@ -92,6 +197,11 @@ public class SyncSwitch { if (mutex != null) { mutex.get(timeout, unit); } + } else { + BooleanMutex mutex = DISTRIBUTED_LOCK.get(destination); + if (mutex != null) { + mutex.get(timeout, unit); + } } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java index 3dcfe4e7..18a12d7c 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java @@ -13,7 +13,7 @@ import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; @ConfigurationProperties(prefix = "canal.conf") public class AdapterCanalConfig extends CanalClientConfig { - public static final Set DESTINATIONS = new LinkedHashSet<>(); + public final Set DESTINATIONS = new LinkedHashSet<>(); @Override public void setCanalInstances(List canalInstances) { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/CuratorClient.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/CuratorClient.java new file mode 100644 index 00000000..b7944852 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/CuratorClient.java @@ -0,0 +1,36 @@ +package com.alibaba.otter.canal.adapter.launcher.config; + +import javax.annotation.PostConstruct; +import javax.annotation.Resource; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.ExponentialBackoffRetry; +import org.springframework.stereotype.Component; + +@Component +public class CuratorClient { + + @Resource + private AdapterCanalConfig adapterCanalConfig; + + private CuratorFramework curator = null; + + @PostConstruct + public void init() { + if (adapterCanalConfig.getZookeeperHosts() != null) { + curator = CuratorFrameworkFactory.builder() + .connectString(adapterCanalConfig.getZookeeperHosts()) + .retryPolicy(new ExponentialBackoffRetry(1000, 3)) + .sessionTimeoutMs(6000) + .connectionTimeoutMs(3000) + .namespace("canal-adapter") + .build(); + curator.start(); + } + } + + public CuratorFramework getCurator() { + return curator; + } +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index d9cdaa7c..3bb5b4a4 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -15,8 +15,8 @@ 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; +import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; /** * 外部适配器的加载器 @@ -66,10 +66,12 @@ public class CanalAdapterLoader { canalOuterAdapterGroups.add(canalOutConnectors); } CanalAdapterWorker worker; - if (zkHosts != null) { + if (sa != null) { + worker = new CanalAdapterWorker(instance.getInstance(), sa, canalOuterAdapterGroups); + } else if (zkHosts != null) { worker = new CanalAdapterWorker(instance.getInstance(), zkHosts, canalOuterAdapterGroups); } else { - worker = new CanalAdapterWorker(instance.getInstance(), sa, canalOuterAdapterGroups); + throw new RuntimeException("No canal server connector found"); } canalWorkers.put(instance.getInstance(), worker); worker.start(); diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java index b976bb82..1c324996 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java @@ -54,8 +54,6 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); connector = CanalConnectors.newClusterConnector(zookeeperHosts, canalDestination, "", ""); ((ClusterCanalConnector) connector).setSoTimeout(SO_TIMEOUT); - - // super.initSwitcher(canalDestination); } @Override diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java index 61c2b47b..d08af780 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java @@ -26,6 +26,9 @@ public class CommonRest { @Resource private SyncSwitch syncSwitch; + @Resource + private AdapterCanalConfig adapterCanalConfig; + @PostConstruct public void init() { loader = ExtensionLoader.getExtensionLoader(OuterAdapter.class); @@ -82,7 +85,7 @@ public class CommonRest { @GetMapping("/destinations") public List> destinations() { List> result = new ArrayList<>(); - Set destinations = AdapterCanalConfig.DESTINATIONS; + Set destinations = adapterCanalConfig.DESTINATIONS; for (String destination : destinations) { Map resMap = new LinkedHashMap<>(); Boolean status = syncSwitch.status(destination); diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 83ccb66d..9030cb71 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -16,14 +16,14 @@ hbase.zookeeper.znode.parent: /hbase-unsecure canal.conf: canalServerHost: 127.0.0.1:11111 -# zookeeperHosts: slave1:2181 + zookeeperHosts: slave1:2181 # bootstrapServers: slave1:6667 #or rocketmq nameservers:host1:9876;host2:9876 flatMessage: true canalInstances: - instance: example adapterGroups: - outAdapters: - - name: logger +# - name: logger - name: hbase properties: hbase.zookeeper.quorum: ${hbase.zookeeper.quorum} @@ -45,4 +45,3 @@ adapter.conf: password: 121212 adapterConfigs: - hbase/mytest_person2.yml -# - hbase/mytest.person2@defaultDS