diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/IndexDataCapacityMonitor.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/IndexDataCapacityMonitor.java index 66e827d80..608ee39af 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/IndexDataCapacityMonitor.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/IndexDataCapacityMonitor.java @@ -2,6 +2,7 @@ package com.a.eye.skywalking.storage.data; import com.a.eye.skywalking.storage.block.index.BlockIndexEngine; import com.a.eye.skywalking.storage.data.index.IndexDBConnector; +import org.apache.jute.Index; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -13,10 +14,9 @@ import static com.a.eye.skywalking.storage.config.Config.DataIndex.MAX_CAPACITY_ /** * Created by xin on 2016/11/6. */ -public class IndexDataCapacityMonitor { +public class IndexDataCapacityMonitor extends Thread { private static Logger logger = LogManager.getLogger(IndexDataCapacityMonitor.class); - private static Detector detector; public static void addIndexData(long timestamp, int size) { @@ -25,7 +25,7 @@ public class IndexDataCapacityMonitor { } } - private static class Detector { + private class Detector { private AtomicLong currentSize; private long timestamp; @@ -46,21 +46,27 @@ public class IndexDataCapacityMonitor { public void add(int updateRecordSize) { if (currentSize.addAndGet(updateRecordSize) > MAX_CAPACITY_PER_INDEX * 0.8) { - notificationAddNewBlockIndex(); + notificationAddNewBlockIndexAndCreateNewIndexDB(); } } } - private static void notificationAddNewBlockIndex() { + private void notificationAddNewBlockIndexAndCreateNewIndexDB() { long timestamp = System.currentTimeMillis() + 5 * 60 * 1000; BlockIndexEngine.newUpdator().addRecord(timestamp); + createNewIndexDB(timestamp); detector = new Detector(timestamp); } - public static void start() { + private void createNewIndexDB(long timestamp) { + IndexDBConnector connector = new IndexDBConnector(timestamp); + connector.close(); + } + + @Override + public void run() { long timestamp = BlockIndexEngine.newFinder().findLastBlockIndex(); - //TODO: 为IndexDBConnector增加闭包执行函数 IndexDBConnector dbConnector = null; try { dbConnector = new IndexDBConnector(timestamp); @@ -72,9 +78,10 @@ public class IndexDataCapacityMonitor { } detector = new Detector(timestamp, count); } finally { - if(dbConnector != null){ + if (dbConnector != null) { dbConnector.close(); } } } + } diff --git a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java index 272663be0..c06e3093c 100644 --- a/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java +++ b/skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexDBConnector.java @@ -137,7 +137,11 @@ public class IndexDBConnector { } } - public void close(){ - //TODO: + public void close() { + try { + connection.close(); + } catch (SQLException e) { + logger.error("Failed to close index db connector", e); + } } }