From beed10901d210d5fc6ad45cebeca2c0df3f27285 Mon Sep 17 00:00:00 2001 From: wuwo Date: Mon, 5 Nov 2018 16:22:15 +0800 Subject: [PATCH 1/2] destory TableMetaTSDB when stop EventParser --- .../main/resources/spring/tsdb/h2-tsdb.xml | 2 +- .../main/resources/spring/tsdb/mysql-tsdb.xml | 2 +- .../inbound/mysql/tsdb/DatabaseTableMeta.java | 22 ++++++++++++++++++- .../inbound/mysql/tsdb/MemoryTableMeta.java | 5 +++++ .../inbound/mysql/tsdb/TableMetaTSDB.java | 5 +++++ parse/src/test/resources/tsdb/h2-tsdb.xml | 2 +- parse/src/test/resources/tsdb/mysql-tsdb.xml | 2 +- 7 files changed, 35 insertions(+), 5 deletions(-) diff --git a/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml b/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml index beb61236..c05b5200 100644 --- a/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml +++ b/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml @@ -23,7 +23,7 @@ - + diff --git a/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml b/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml index 58cab638..87f407b3 100644 --- a/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml +++ b/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml @@ -23,7 +23,7 @@ - + diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java index b0dd89ce..73a5ead8 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java @@ -72,7 +72,7 @@ public class DatabaseTableMeta implements TableMetaTSDB { @Override public Thread newThread(Runnable r) { - Thread thread = new Thread(r, "[scheduler-table-meta-snapshot]"); + Thread thread = new Thread(r, String.format("[scheduler-table-meta-snapshot-%s]", destination)); thread.setDaemon(true); return thread; } @@ -105,6 +105,26 @@ public class DatabaseTableMeta implements TableMetaTSDB { } return true; } + + @Override + public void destory() { + if (memoryTableMeta != null) { + memoryTableMeta.destory(); + } + + if (connection != null) { + try { + connection.disconnect(); + } catch (IOException e) { + logger.error("ERROR # disconnect meta connection for address:{}", connection.getConnector() + .getAddress(), e); + } + } + + if (scheduler != null) { + scheduler.shutdown(); + } + } @Override public TableMeta find(String schema, String table) { diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java index 904b55a9..c53e1a08 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java @@ -58,6 +58,11 @@ public class MemoryTableMeta implements TableMetaTSDB { public boolean init(String destination) { return true; } + + @Override + public void destory() { + tableMetas.clear(); + } public boolean apply(EntryPosition position, String schema, String ddl, String extra) { tableMetas.clear(); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java index d712adbe..f968ae12 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java @@ -18,6 +18,11 @@ public interface TableMetaTSDB { */ public boolean init(String destination); + /** + * 销毁资源 + */ + public void destory(); + /** * 获取当前的表结构 */ diff --git a/parse/src/test/resources/tsdb/h2-tsdb.xml b/parse/src/test/resources/tsdb/h2-tsdb.xml index c0a3d385..dcb6bc47 100644 --- a/parse/src/test/resources/tsdb/h2-tsdb.xml +++ b/parse/src/test/resources/tsdb/h2-tsdb.xml @@ -7,7 +7,7 @@ default-autowire="byName"> - + diff --git a/parse/src/test/resources/tsdb/mysql-tsdb.xml b/parse/src/test/resources/tsdb/mysql-tsdb.xml index 099c034d..44f97203 100644 --- a/parse/src/test/resources/tsdb/mysql-tsdb.xml +++ b/parse/src/test/resources/tsdb/mysql-tsdb.xml @@ -7,7 +7,7 @@ default-autowire="byName"> - + From e608f09b53e1d0d0964fda73693320f3f66148dc Mon Sep 17 00:00:00 2001 From: wuwo Date: Mon, 5 Nov 2018 16:37:42 +0800 Subject: [PATCH 2/2] improve DatabaseTableMeta --- .../inbound/mysql/tsdb/DatabaseTableMeta.java | 28 +++++++++---------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java index 73a5ead8..8896e2f5 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java @@ -7,6 +7,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.regex.Pattern; @@ -48,18 +49,26 @@ public class DatabaseTableMeta implements TableMetaTSDB { private static Logger logger = LoggerFactory.getLogger(DatabaseTableMeta.class); private static Pattern pattern = Pattern.compile("Duplicate entry '.*' for key '*'"); private static Pattern h2Pattern = Pattern.compile("Unique index or primary key violation"); + private static ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() { + @Override + public Thread newThread(Runnable r) { + Thread thread = new Thread(r, "[scheduler-table-meta-snapshot]"); + thread.setDaemon(true); + return thread; + } + }); private String destination; private MemoryTableMeta memoryTableMeta; private MysqlConnection connection; // 查询meta信息的链接 private CanalEventFilter filter; private CanalEventFilter blackFilter; private EntryPosition lastPosition; - private ScheduledExecutorService scheduler; private MetaHistoryDAO metaHistoryDAO; private MetaSnapshotDAO metaSnapshotDAO; private int snapshotInterval = 24; private int snapshotExpire = 360; - + private ScheduledFuture scheduleSnapshotFuture; + public DatabaseTableMeta(){ } @@ -68,19 +77,10 @@ public class DatabaseTableMeta implements TableMetaTSDB { public boolean init(final String destination) { this.destination = destination; this.memoryTableMeta = new MemoryTableMeta(); - this.scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() { - - @Override - public Thread newThread(Runnable r) { - Thread thread = new Thread(r, String.format("[scheduler-table-meta-snapshot-%s]", destination)); - thread.setDaemon(true); - return thread; - } - }); // 24小时生成一份snapshot if (snapshotInterval > 0) { - scheduler.scheduleWithFixedDelay(new Runnable() { + scheduleSnapshotFuture = scheduler.scheduleWithFixedDelay(new Runnable() { @Override public void run() { @@ -121,8 +121,8 @@ public class DatabaseTableMeta implements TableMetaTSDB { } } - if (scheduler != null) { - scheduler.shutdown(); + if (scheduleSnapshotFuture != null) { + scheduleSnapshotFuture.cancel(false); } }