From fcdf95b5ecc3dc59e0d52f3de5fb594162cc6215 Mon Sep 17 00:00:00 2001 From: mcy Date: Sat, 17 Nov 2018 04:37:46 +0800 Subject: [PATCH] =?UTF-8?q?rdb=20=E6=89=B9=E9=87=8F=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=E8=B6=85=E6=97=B6=E6=8F=90=E4=BA=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/client/adapter/es/ESAdapter.java | 7 +++++-- .../adapter/es/service/ESSyncService.java | 1 - .../canal/client/adapter/rdb/RdbAdapter.java | 21 ++++++++----------- 3 files changed, 14 insertions(+), 15 deletions(-) diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java index 18cc0c93..35df8a76 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java @@ -1,14 +1,16 @@ package com.alibaba.otter.canal.client.adapter.es; import java.net.InetAddress; -import java.util.*; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.regex.Matcher; import java.util.regex.Pattern; import javax.sql.DataSource; -import com.alibaba.otter.canal.client.adapter.es.monitor.ESConfigMonitor; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.common.settings.Settings; @@ -22,6 +24,7 @@ import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem; import com.alibaba.otter.canal.client.adapter.es.config.SqlParser; +import com.alibaba.otter.canal.client.adapter.es.monitor.ESConfigMonitor; import com.alibaba.otter.canal.client.adapter.es.service.ESEtlService; import com.alibaba.otter.canal.client.adapter.es.service.ESSyncService; import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate; diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java index 397e3e5a..823a9d4c 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java @@ -14,7 +14,6 @@ import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; -import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.ColumnItem; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem; diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java index 78001c03..291ecbab 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java @@ -5,7 +5,6 @@ import java.sql.SQLException; import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -47,7 +46,6 @@ public class RdbAdapter implements OuterAdapter { private List dmlList = Collections .synchronizedList(new ArrayList<>()); private Lock syncLock = new ReentrantLock(); - private Condition condition = syncLock.newCondition(); private ExecutorService executor = Executors.newFixedThreadPool(1); private RdbConfigMonitor rdbConfigMonitor; @@ -112,16 +110,16 @@ public class RdbAdapter implements OuterAdapter { executor.submit(() -> { while (running) { + int beginSize = dmlList.size(); try { - syncLock.lock(); - if (!condition.await(3, TimeUnit.SECONDS)) { - // 超时提交 - sync(); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } finally { - syncLock.unlock(); + Thread.sleep(3000); + } catch (InterruptedException e) { + // ignore + } + int endSize = dmlList.size(); + + if (endSize - beginSize < 300) { + sync(); } } }); @@ -156,7 +154,6 @@ public class RdbAdapter implements OuterAdapter { try { syncLock.lock(); if (!dmlList.isEmpty()) { - condition.signal(); rdbSyncService.sync(dmlList); dmlList.clear(); }