add retry scheduler name

Author: 胡贵

README.md

@@ -54,7 +54,7 @@ LTS 轻量级分布式任务调度框架(Light Task Schedule)
 ###LTS Admin
 ![Aaron Swartz](https://raw.githubusercontent.com/qq254963746/light-task-schedule/master/doc/LTS_Admin.png)
 ###调用示例
-下面提供的是最简单的配置方式。更多配置请查看 [lts-example](https://github.com/qq254963746/light-task-schedule/tree/master/lts-example/src/main/java/com/lts/job/example/api) 模块下的 API 调用方式例子.
+下面提供的是最简单的配置方式。更多配置请查看 [lts-example](https://github.com/qq254963746/light-task-schedule/tree/master/lts-example/src/main/java/com/lts/example/api) 模块下的 API 调用方式例子.
 
 ####JobTracker 端
 ```java
@@ -85,7 +85,7 @@ LTS 轻量级分布式任务调度框架(Light Task Schedule)
         public void run(Job job) throws Throwable {
             System.out.println("我要执行"+ job);
             System.out.println(job.getParam("shopId"));
-            // TODO 用户自己的业务逻辑
+            // TODO 用户自己的业务逻辑, 应该保证幂等
             try {
                 Thread.sleep(5*1000L);
             } catch (InterruptedException e) {

lts-core/src/main/java/com/lts/core/failstore/leveldb/LeveldbFailStore.java

@@ -7,12 +7,11 @@
 import com.lts.core.failstore.FailStore;
 import com.lts.core.failstore.FailStoreException;
 import org.fusesource.leveldbjni.JniDBFactory;
-import org.iq80.leveldb.DB;
-import org.iq80.leveldb.DBIterator;
-import org.iq80.leveldb.Options;
+import org.iq80.leveldb.*;
 
 import java.io.File;
 import java.io.IOException;
+import java.io.UnsupportedEncodingException;
 import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.List;
@@ -39,13 +38,22 @@ public LeveldbFailStore(String failStorePath) {
         failStorePath = failStorePath + "/leveldb/";
         dbPath = FileUtils.createDirIfNotExist(failStorePath);
         options = new Options();
+        options.createIfMissing(true);
+        options.cacheSize(100 * 1024 * 1024);   // 100M
+//        options.logger(new Logger() {
+//            @Override
+//            public void log(String message) {
+//                System.out.println(message);
+//            }
+//        });
         lock = new FileLock(failStorePath + "___db.lock");
     }
 
     @Override
     public void open() throws FailStoreException {
         try {
             lock.tryLock();
+            JniDBFactory.factory.repair(dbPath, options);
             db = JniDBFactory.factory.open(dbPath, options);
         } catch (IOException e) {
             throw new FailStoreException(e);
@@ -79,16 +87,30 @@ public void delete(List<String> keys) throws FailStoreException {
         if (keys == null || keys.size() == 0) {
             return;
         }
-        for (String key : keys) {
-            delete(key);
+        WriteBatch batch = db.createWriteBatch();
+        try {
+
+            for (String key : keys) {
+                batch.delete(key.getBytes("UTF-8"));
+            }
+            db.write(batch);
+        } catch (UnsupportedEncodingException e) {
+            throw new FailStoreException(e);
+        } finally {
+            try {
+                batch.close();
+            } catch (IOException e) {
+                throw new FailStoreException(e);
+            }
         }
     }
 
     @Override
     public <T> List<KVPair<String, T>> fetchTop(int size, Type type) throws FailStoreException {
+        DBIterator iterator = null;
         try {
             List<KVPair<String, T>> list = new ArrayList<KVPair<String, T>>(size);
-            DBIterator iterator = db.iterator();
+            iterator = db.iterator();
             for (iterator.seekToFirst(); iterator.hasNext(); iterator.next()) {
                 Map.Entry<byte[], byte[]> entry = iterator.peekNext();
                 String key = new String(entry.getKey(), "UTF-8");
@@ -102,6 +124,14 @@ public <T> List<KVPair<String, T>> fetchTop(int size, Type type) throws FailStor
             return list;
         } catch (Exception e) {
             throw new FailStoreException(e);
+        } finally {
+            if (iterator != null) {
+                try {
+                    iterator.close();
+                } catch (IOException e) {
+                    throw new FailStoreException(e);
+                }
+            }
         }
     }
 

lts-core/src/main/java/com/lts/core/failstore/rocksdb/RocksdbFailStore.java

@@ -103,7 +103,7 @@ public <T> List<KVPair<String, T>> fetchTop(int size, Type type) throws FailStor
         try {
             List<KVPair<String, T>> list = new ArrayList<KVPair<String, T>>(size);
             iterator = db.newIterator();
-            for (iterator.seekToLast(); iterator.isValid(); iterator.prev()) {
+            for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) {
                 iterator.status();
                 String key = new String(iterator.key(), "UTF-8");
                 T value = JSONUtils.parse(new String(iterator.value(), "UTF-8"), type);

lts-core/src/test/java/com/lts/core/failstore/berkeleydb/BerkeleydbFailStoreTest.java

@@ -37,10 +37,11 @@ public void setup() throws FailStoreException {
     public void put() throws FailStoreException {
         Job job = new Job();
         job.setTaskId("2131232");
-        failStore.put(key, job);
-        failStore.close();
+        for (int i = 0; i < 100; i++) {
+            failStore.put(key + "" + i, job);
+        }
         System.out.println("这里debug测试多线程");
-//        fetchTop();
+        failStore.close();
     }
 
     @Test

lts-core/src/test/java/com/lts/core/failstore/leveldb/LeveldbFailStoreTest.java

@@ -32,7 +32,9 @@ public void setup() throws FailStoreException {
     public void put() throws FailStoreException {
         Job job = new Job();
         job.setTaskId("2131232");
-        failStore.put(key, job);
+        for (int i = 0; i < 100; i++) {
+            failStore.put(key + "" + i, job);
+        }
         System.out.println("这里debug测试多线程");
         failStore.close();
     }

lts-core/src/test/java/com/lts/core/failstore/rocksdb/RocksdbFailStoreTest.java

@@ -37,7 +37,9 @@ public void setup() throws FailStoreException {
     public void put() throws FailStoreException {
         Job job = new Job();
         job.setTaskId("2131232");
-        failStore.put(key, job);
+        for (int i = 0; i < 100; i++) {
+            failStore.put(key + "" + i, job);
+        }
         System.out.println("这里debug测试多线程");
         failStore.close();
     }