cc 2 сар өмнө
parent
commit
a27ec0b5c3
21 өөрчлөгдсөн 143 нэмэгдсэн , 1093 устгасан
  1. 6 7
      pom.xml
  2. 2 0
      src/main/java/com/qingjian/App.java
  3. 13 0
      src/main/java/com/qingjian/common/cache/CaseDataCache.java
  4. 11 37
      src/main/java/com/qingjian/common/cache/GlobalCache.java
  5. 5 1
      src/main/java/com/qingjian/common/cache/RocksDBCache.java
  6. 45 0
      src/main/java/com/qingjian/common/cache/RogueBaseDataCache.java
  7. 2 3
      src/main/java/com/qingjian/common/config/WebConfig.java
  8. 2 0
      src/main/java/com/qingjian/common/constants/PathConst.java
  9. 1 14
      src/main/java/com/qingjian/common/event/AppLoadEndEventListener.java
  10. 4 3
      src/main/java/com/qingjian/common/event/AppStopEndEventListener.java
  11. 9 16
      src/main/java/com/qingjian/common/model/person/entity/PersonLibNo.java
  12. 0 593
      src/main/java/com/qingjian/common/rocksdb/AbstractRocksDBStorage.java
  13. 0 205
      src/main/java/com/qingjian/common/rocksdb/DefaultRocksDBStorage.java
  14. 0 173
      src/main/java/com/qingjian/common/rocksdb/RocksDBHelper.java
  15. 5 5
      src/main/java/com/qingjian/common/utils/DataUtil.java
  16. 10 18
      src/main/java/com/qingjian/module/dm/clean/AbstractCallRecordDataLoader.java
  17. 3 1
      src/main/java/com/qingjian/module/dm/clean/AbstractDataLoader.java
  18. 8 14
      src/main/java/com/qingjian/module/dm/clean/AbstractTransRecordDataLoader.java
  19. 1 3
      src/main/java/com/qingjian/module/person/controller/DataProfileController.java
  20. 14 0
      src/main/java/com/qingjian/module/plat/service/SseService.java
  21. 2 0
      src/test/java/com/qingjian/common/cache/RocksDBCacheTest.java

+ 6 - 7
pom.xml

@@ -58,6 +58,11 @@
             <artifactId>solon-view-thymeleaf</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.noear</groupId>
+            <artifactId>solon-scheduling-simple</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.noear</groupId>
             <artifactId>solon-web-staticfiles</artifactId>
@@ -146,12 +151,6 @@
             <version>7.0.2</version>
         </dependency>
 
-        <dependency>
-            <groupId>org.rocksdb</groupId>
-            <artifactId>rocksdbjni</artifactId>
-            <version>10.2.1</version>
-        </dependency>
-
         <dependency>
             <groupId>cn.hutool</groupId>
             <artifactId>hutool-all</artifactId>
@@ -233,7 +232,7 @@
         <dependency>
             <groupId>com.yomahub</groupId>
             <artifactId>roguemap-core</artifactId>
-            <version>1.1.3</version>
+            <version>1.1.7</version>
         </dependency>
 
         <!-- ik分词器 -->

+ 2 - 0
src/main/java/com/qingjian/App.java

@@ -5,10 +5,12 @@ import com.qingjian.common.constants.PathConst;
 import lombok.extern.slf4j.Slf4j;
 import org.noear.solon.Solon;
 import org.noear.solon.annotation.SolonMain;
+import org.noear.solon.scheduling.annotation.EnableScheduling;
 import org.noear.solon.web.cors.CrossFilter;
 
 import java.nio.file.Files;
 
+@EnableScheduling
 @SolonMain
 @Slf4j
 public class App {

+ 13 - 0
src/main/java/com/qingjian/common/cache/CaseDataCache.java

@@ -36,6 +36,8 @@ public class CaseDataCache {
      */
     public static final HashMultimap<String, String> PERSON_LIB_NO_MAP = HashMultimap.create();
     public static final Map<String, String> GOVERN_CONFIG = Maps.newConcurrentMap();
+    public static final Map<String, String> PERSON_CARD = Maps.newConcurrentMap();
+    public static final Map<String, String> PERSON_PHONE = Maps.newConcurrentMap();
     // 使用 ReentrantReadWriteLock 替代单一的 Object 锁
     private static final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
     private static final ReentrantReadWriteLock.ReadLock readLock = rwLock.readLock();
@@ -53,6 +55,11 @@ public class CaseDataCache {
         personLibNos.forEach(personLibNo -> {
             String key = personLibNo.getPersonName() + StrPool.UNDERLINE + personLibNo.getLibType();
             PERSON_LIB_NO_MAP.put(key, personLibNo.getPersonNo());
+            if (personLibNo.getLibType() == 2) {
+                PERSON_PHONE.put(personLibNo.getPersonNo(), personLibNo.getPersonName());
+            } else if (personLibNo.getLibType() == 3) {
+                PERSON_CARD.put(personLibNo.getPersonNo(), personLibNo.getPersonName());
+            }
         });
     }
 
@@ -65,6 +72,12 @@ public class CaseDataCache {
         try {
             PERSON_LIB_NO_MAP.put(key, value);
             String[] keys = StrUtil.splitToArray(key, StrPool.UNDERLINE);
+            int i = Integer.parseInt(keys[1]);
+            if (i == 2) {
+                PERSON_PHONE.put(value, keys[0]);
+            } else if (i == 3) {
+                PERSON_CARD.put(value, keys[0]);
+            }
             CleanCache.addPersonLibNo(new PersonLibNo(keys[0], value, Integer.parseInt(keys[1]), 1, bankName));
         } finally {
             // 必须在 finally 块中释放锁,确保即使发生异常也能正确释放

+ 11 - 37
src/main/java/com/qingjian/common/cache/GlobalCache.java

@@ -1,6 +1,7 @@
 package com.qingjian.common.cache;
 
 import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.convert.Convert;
 import cn.hutool.core.lang.PatternPool;
 import cn.hutool.core.text.StrPool;
 import cn.hutool.core.util.StrUtil;
@@ -14,9 +15,6 @@ import com.qingjian.common.model.plat.entity.CardIssuerBank;
 import com.qingjian.common.model.plat.entity.PhoneIsp;
 import com.qingjian.common.model.plat.entity.TableField;
 import com.qingjian.common.model.plat.entity.TableInfo;
-import com.qingjian.common.rocksdb.DefaultRocksDBStorage;
-import com.qingjian.common.rocksdb.RocksDBHelper;
-import com.qingjian.common.utils.DataUtil;
 import com.qingjian.common.utils.Json;
 import com.qingjian.common.utils.SystemConfManager;
 import com.qingjian.module.plat.mapper.CardIssuerBankMapper;
@@ -26,8 +24,6 @@ import com.qingjian.module.plat.mapper.TableInfoMapper;
 import lombok.extern.slf4j.Slf4j;
 import org.noear.solon.Solon;
 import org.noear.solon.data.dynamicds.DynamicDsKey;
-import org.rocksdb.RocksDBException;
-import org.rocksdb.WriteBatch;
 
 import java.time.LocalDateTime;
 import java.util.Comparator;
@@ -274,21 +270,18 @@ public class GlobalCache {
     }
 
     public static void initCaseRocksDbData() {
-        String bytes = RocksDBHelper.INSTANCE.getIspValue(DataUtil.getPhonePrefix7("1300012"));
+        String bytes = RogueBaseDataCache.INSTANCE.get(1300012);
         log.info("Test RockDB IspData 1300012={}", bytes);
         if (SystemConfManager.INSTANCE.getInitFlag()) {
             return;
         }
         SystemConfManager.INSTANCE.setInitFlag(true);
+        log.info("开始初始化全局缓存...");
         CardIssuerBankMapper cardIssuerBankMapper = Solon.context().getBean(CardIssuerBankMapper.class);
         List<CardIssuerBank> list = DynamicDsKey.with(StrConsts.DS_KEY_PLAT, () -> cardIssuerBankMapper.selectList(Wrappers.emptyWrapper()));
-        Map<String, String> bankBinMap = list.stream()
-                .collect(Collectors.toMap(i -> DataUtil.getCardPrefix6(i.getBankBin()), CardIssuerBank::getBankName, (_, k2) -> k2));
-        try {
-            RocksDBHelper.INSTANCE.batchPut(bankBinMap);
-        } catch (RocksDBException e) {
-            log.error("RocksDB data init CardIssuerBank failed", e);
-        }
+        Map<Integer, String> bankBinMap = list.stream()
+                .collect(Collectors.toMap(i -> Convert.toInt(i.getBankBin()), CardIssuerBank::getBankName, (_, k2) -> k2));
+        RogueBaseDataCache.INSTANCE.putAll(bankBinMap);
         long l = System.currentTimeMillis();
         initIspData();
         log.info("initIspData cost:{}", System.currentTimeMillis() - l);
@@ -299,30 +292,11 @@ public class GlobalCache {
      */
     private static void initIspData() {
         PhoneIspMapper ispMapper = Solon.context().getBean(PhoneIspMapper.class);
-        try (WriteBatch batch = new WriteBatch()) {
-            List<PhoneIsp> isps = DynamicDsKey.with(StrConsts.DS_KEY_PLAT, () -> ispMapper.selectList(Wrappers.emptyWrapper()));
-            // 每批次10万条
-            final int batchSize = 100_000;
-            DefaultRocksDBStorage storage = RocksDBHelper.INSTANCE.get();
-
-            for (int i = 0; i < isps.size(); i++) {
-                PhoneIsp phoneIsp = isps.get(i);
-                String paragraph = DataUtil.getPhonePrefix7(phoneIsp.getParagraph());
-                String isp = phoneIsp.getIsp();
-                byte[] key = StrUtil.utf8Bytes(paragraph);
-                byte[] value = StrUtil.utf8Bytes(isp);
-                batch.put(storage.ipsDataFamilyHandle, key, value);
-                // 分批提交
-                if ((i + 1) % batchSize == 0) {
-                    storage.batchIpsPut(batch);
-                }
-            }
-            // 提交剩余数据
-            if (batch.count() > 0) {
-                storage.batchIpsPut(batch);
-            }
-        } catch (RocksDBException e) {
-            log.error("RocksDB data init IspData failed", e);
+        List<PhoneIsp> isps = DynamicDsKey.with(StrConsts.DS_KEY_PLAT, () -> ispMapper.selectList(Wrappers.emptyWrapper()));
+        for (PhoneIsp phoneIsp : isps) {
+            int paragraph = Convert.toInt(phoneIsp.getParagraph());
+            String isp = phoneIsp.getIsp();
+            RogueBaseDataCache.INSTANCE.put(paragraph, isp);
         }
     }
 }

+ 5 - 1
src/main/java/com/qingjian/common/cache/RocksDBCache.java

@@ -1,3 +1,4 @@
+/*
 package com.qingjian.common.cache;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -11,12 +12,14 @@ import java.lang.reflect.Type;
 import java.nio.charset.StandardCharsets;
 import java.util.function.Supplier;
 
+*/
 /**
  * 基于 RocksDB 列簇的 Solon 缓存服务。
  *
  * @author cc
  * @since 2026/4/22
- */
+ *//*
+
 @Slf4j
 public class RocksDBCache implements CacheService {
 
@@ -116,3 +119,4 @@ public class RocksDBCache implements CacheService {
 
     }
 }
+*/

+ 45 - 0
src/main/java/com/qingjian/common/cache/RogueBaseDataCache.java

@@ -0,0 +1,45 @@
+package com.qingjian.common.cache;
+
+import com.qingjian.common.constants.PathConst;
+import com.yomahub.roguemap.RogueMap;
+import com.yomahub.roguemap.serialization.PrimitiveCodecs;
+import com.yomahub.roguemap.serialization.StringCodec;
+
+import java.util.Map;
+
+public class RogueBaseDataCache {
+
+    public static final RogueBaseDataCache INSTANCE = new RogueBaseDataCache();
+
+    private final RogueMap<Integer, String> ROGUE_CACHE;
+
+    public RogueBaseDataCache() {
+        ROGUE_CACHE = RogueMap.<Integer, String>mmap()
+                .persistent(PathConst.SYSTEM_ROGUE_PATH.toString())
+                .autoExpand(true)
+                .keyCodec(PrimitiveCodecs.INTEGER)
+                .valueCodec(StringCodec.INSTANCE)
+                .primitiveIndex()
+                .build();
+    }
+
+    public String get(Integer key) {
+        return ROGUE_CACHE.get(key);
+    }
+
+    public void put(Integer key, String value) {
+        ROGUE_CACHE.put(key, value);
+    }
+
+    public void putAll(Map<Integer, String> data) {
+        ROGUE_CACHE.putAll(data);
+    }
+
+    public void flush() {
+        ROGUE_CACHE.flush();
+    }
+
+    public void close() {
+        ROGUE_CACHE.close();
+    }
+}

+ 2 - 3
src/main/java/com/qingjian/common/config/WebConfig.java

@@ -1,8 +1,6 @@
 package com.qingjian.common.config;
 
-import com.qingjian.common.cache.RocksDBCache;
 import com.qingjian.common.constants.PathConst;
-import com.qingjian.common.rocksdb.RocksDBHelper;
 import com.qingjian.common.utils.Json;
 import org.noear.solon.annotation.Bean;
 import org.noear.solon.annotation.Configuration;
@@ -10,6 +8,7 @@ import org.noear.solon.annotation.Inject;
 import org.noear.solon.data.cache.CacheService;
 import org.noear.solon.data.cache.CacheTagsService;
 import org.noear.solon.data.cache.CacheTagsServiceImpl;
+import org.noear.solon.data.cache.LocalCacheService;
 import org.noear.solon.data.dynamicds.DynamicDataSource;
 import org.noear.solon.serialization.jackson.JacksonStringSerializer;
 
@@ -39,7 +38,7 @@ public class WebConfig {
 
     @Bean
     public CacheService cache() {
-        return new RocksDBCache(Json.objectMapper(), RocksDBHelper.INSTANCE::get);
+        return new LocalCacheService(21600000);
     }
 
     @Bean

+ 2 - 0
src/main/java/com/qingjian/common/constants/PathConst.java

@@ -45,10 +45,12 @@ public class PathConst {
      * 系统内置 银行卡和银行名称、手机号和运营商关系 的 rocksdb名称
      */
     public static final String SYSTEM_ROCKSDB_NAME = "d5e03ad1";
+    public static final String SYSTEM_ROGUE_NAME = "uy76tkp8";
     /**
      * 系统内置 银行卡和银行名称、手机号和运营商关系 的 rocksdb路径
      */
     public static final Path SYSTEM_ROCKSDB_PATH = RUN_QINGJIAN_PATH.resolve(SYSTEM_ROCKSDB_NAME);
+    public static final Path SYSTEM_ROGUE_PATH = RUN_QINGJIAN_PATH.resolve(SYSTEM_ROGUE_NAME,"data.db");
     /**
      * 基础案件数据库路径
      */

+ 1 - 14
src/main/java/com/qingjian/common/event/AppLoadEndEventListener.java

@@ -4,8 +4,6 @@ import cn.hutool.core.lang.Singleton;
 import com.qingjian.common.cache.CaseDataCache;
 import com.qingjian.common.cache.GlobalCache;
 import com.qingjian.common.constants.PathConst;
-import com.qingjian.common.rocksdb.DefaultRocksDBStorage;
-import com.qingjian.common.rocksdb.RocksDBHelper;
 import com.qingjian.common.utils.TowerUtils;
 import com.qingjian.module.ai.agent.AgentRuntime;
 import com.qingjian.module.ai.service.AiService;
@@ -16,7 +14,6 @@ import org.noear.solon.core.event.AppLoadEndEvent;
 import org.noear.solon.core.event.EventListener;
 
 import java.nio.file.Files;
-import java.nio.file.Path;
 
 @Component
 @Slf4j
@@ -29,8 +26,8 @@ public class AppLoadEndEventListener implements EventListener<AppLoadEndEvent> {
     public void onEvent(AppLoadEndEvent event) {
         //event.app(); //获取应用对象
         initDir();
-        initRocks();
         GlobalCache.initCache();
+        GlobalCache.initCaseRocksDbData();
         log.info("初始化缓存配置完成...");
         CaseDataCache.initCache();
         log.info("初始化缓存完成...");
@@ -68,15 +65,5 @@ public class AppLoadEndEventListener implements EventListener<AppLoadEndEvent> {
         }
     }
 
-    private void initRocks() {
-        Path path = PathConst.SYSTEM_ROCKSDB_PATH;
-        DefaultRocksDBStorage storage = new DefaultRocksDBStorage(path.toString());
-        if (storage.start()) {
-            RocksDBHelper.INSTANCE.put(storage);
-            GlobalCache.initCaseRocksDbData();
-            log.info("初始化RocksDB完成...");
-        }
-    }
-
 
 }

+ 4 - 3
src/main/java/com/qingjian/common/event/AppStopEndEventListener.java

@@ -1,6 +1,6 @@
 package com.qingjian.common.event;
 
-import com.qingjian.common.rocksdb.RocksDBHelper;
+import com.qingjian.common.cache.RogueBaseDataCache;
 import com.qingjian.common.utils.CellDll;
 import com.qingjian.common.utils.LuceneManager;
 import org.noear.solon.annotation.Component;
@@ -10,10 +10,11 @@ import org.noear.solon.core.event.EventListener;
 @Component
 public class  AppStopEndEventListener implements EventListener<AppStopEndEvent>{
     @Override
-    public void onEvent(AppStopEndEvent event) throws Throwable {
+    public void onEvent(AppStopEndEvent event) {
         //event.app(); //获取应用对象
-        RocksDBHelper.INSTANCE.remove();
         LuceneManager.close();
+        RogueBaseDataCache.INSTANCE.flush();
+        RogueBaseDataCache.INSTANCE.close();
         CellDll.INSTANCE_DLL.cell_fini();
     }
 }

+ 9 - 16
src/main/java/com/qingjian/common/model/person/entity/PersonLibNo.java

@@ -1,18 +1,17 @@
 package com.qingjian.common.model.person.entity;
 
+import cn.hutool.core.convert.Convert;
 import cn.hutool.core.util.StrUtil;
 import cn.hutool.crypto.digest.DigestUtil;
 import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
-import com.qingjian.common.rocksdb.RocksDBHelper;
-import com.qingjian.common.utils.DataUtil;
+import com.qingjian.common.cache.RogueBaseDataCache;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
-import org.rocksdb.RocksDBException;
 
 import java.time.LocalDateTime;
 
@@ -78,24 +77,18 @@ public class PersonLibNo {
     }
 
     private void updatePersonNoCn(String personNo, Integer libType, String operatorId) {
+        Integer key = Convert.toInt(personNo, 0);
+        if (key == 0) {
+            return;
+        }
         if (StrUtil.isNotEmpty(operatorId)) {
             this.personNoCn = operatorId;
-            try {
-                if (libType == 2) {
-                    String phonePrefix7 = DataUtil.getPhonePrefix7(personNo);
-                    RocksDBHelper.INSTANCE.get().updateIspData(StrUtil.utf8Bytes(phonePrefix7), StrUtil.utf8Bytes(operatorId));
-                } else if (libType == 3) {
-                    String cardPrefix6 = DataUtil.getCardPrefix6(personNo);
-                    RocksDBHelper.INSTANCE.put(StrUtil.utf8Bytes(cardPrefix6), StrUtil.utf8Bytes(operatorId));
-                }
-            } catch (RocksDBException e) {
-                log.error("新增RocksDBException:", e);
-            }
+            RogueBaseDataCache.INSTANCE.put(key, operatorId);
         } else {
             if (libType == 2) {
-                this.personNoCn = RocksDBHelper.INSTANCE.getIspValue(DataUtil.getPhonePrefix7(personNo));
+                this.personNoCn = RogueBaseDataCache.INSTANCE.get(key);
             } else if (libType == 3) {
-                String str = RocksDBHelper.INSTANCE.get(DataUtil.getCardPrefix6(personNo));
+                String str = RogueBaseDataCache.INSTANCE.get(key);
                 this.personNoCn = StrUtil.emptyToDefault(str, "未知银行");
             }
         }

+ 0 - 593
src/main/java/com/qingjian/common/rocksdb/AbstractRocksDBStorage.java

@@ -1,593 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package com.qingjian.common.rocksdb;
-
-import cn.hutool.core.map.MapUtil;
-import cn.hutool.core.util.ArrayUtil;
-import lombok.extern.slf4j.Slf4j;
-import org.rocksdb.*;
-import org.slf4j.Logger;
-
-import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.function.BiConsumer;
-@Slf4j
-public abstract class AbstractRocksDBStorage {
-
-    /**
-     * Direct Jemalloc allocator
-     */
-    private static final String SPACE = " | ";
-
-    protected final String dbPath;
-    protected boolean readOnly;
-    protected RocksDB db;
-    protected DBOptions options;
-
-    protected WriteOptions writeOptions;
-    //protected WriteOptions ableWalWriteOptions;
-
-    protected ReadOptions readOptions;
-    protected ReadOptions totalOrderReadOptions;
-
-    protected CompactionOptions compactionOptions;
-    protected CompactRangeOptions compactRangeOptions;
-
-    protected ColumnFamilyHandle defaultCFHandle;
-    protected final List<ColumnFamilyOptions> cfOptions = new ArrayList<>();
-    protected final List<ColumnFamilyHandle> cfHandles = new ArrayList<>();
-
-    protected CompressionType compressionType = CompressionType.LZ4_COMPRESSION;
-    protected volatile boolean loaded;
-    private volatile boolean closed;
-
-    static {
-        RocksDB.loadLibrary();
-    }
-
-    public AbstractRocksDBStorage(String dbPath) {
-        this.dbPath = dbPath;
-    }
-
-    protected void initOptions() {
-        initWriteOptions();
-        //initAbleWalWriteOptions();
-        initReadOptions();
-        initTotalOrderReadOptions();
-        initCompactRangeOptions();
-        initCompactionOptions();
-    }
-
-    /**
-     * Write options for <a href="https://github.com/facebook/rocksdb/wiki/Atomic-flush">Atomic Flush</a>
-     */
-    protected void initWriteOptions() {
-        this.writeOptions = new WriteOptions();
-        this.writeOptions.setSync(false);
-        this.writeOptions.setDisableWAL(true);
-        // https://github.com/facebook/rocksdb/wiki/Write-Stalls
-        this.writeOptions.setNoSlowdown(false);
-    }
-
-/*    protected void initAbleWalWriteOptions() {
-        this.ableWalWriteOptions = new WriteOptions();
-        this.ableWalWriteOptions.setSync(false);
-        this.ableWalWriteOptions.setDisableWAL(false);
-        // https://github.com/facebook/rocksdb/wiki/Write-Stalls
-        this.ableWalWriteOptions.setNoSlowdown(false);
-    }*/
-
-    protected void initReadOptions() {
-        this.readOptions = new ReadOptions();
-        this.readOptions.setPrefixSameAsStart(true);
-        this.readOptions.setTotalOrderSeek(false);
-        this.readOptions.setTailing(false);
-        this.readOptions.setVerifyChecksums(false);
-    }
-
-    protected void initTotalOrderReadOptions() {
-        this.totalOrderReadOptions = new ReadOptions();
-        this.totalOrderReadOptions.setPrefixSameAsStart(false);
-        this.totalOrderReadOptions.setTotalOrderSeek(true);
-        this.totalOrderReadOptions.setTailing(false);
-        this.totalOrderReadOptions.setVerifyChecksums(false);
-    }
-
-    protected void initCompactRangeOptions() {
-        this.compactRangeOptions = new CompactRangeOptions();
-        this.compactRangeOptions.setBottommostLevelCompaction(CompactRangeOptions.BottommostLevelCompaction.kForce);
-        this.compactRangeOptions.setAllowWriteStall(true);
-        this.compactRangeOptions.setExclusiveManualCompaction(false);
-        this.compactRangeOptions.setChangeLevel(true);
-        this.compactRangeOptions.setTargetLevel(-1);
-        this.compactRangeOptions.setMaxSubcompactions(4);
-    }
-
-    protected void initCompactionOptions() {
-        this.compactionOptions = new CompactionOptions();
-        this.compactionOptions.setCompression(compressionType);
-        this.compactionOptions.setMaxSubcompactions(4);
-        this.compactionOptions.setOutputFileSizeLimit(4 * 1024 * 1024 * 1024L);
-    }
-
-    public boolean hold() {
-        if (!this.loaded || this.db == null || this.closed) {
-            log.error("hold rocksdb Failed. {}", this.dbPath);
-            return false;
-        } else {
-            return true;
-        }
-    }
-
-    public void release() {
-    }
-
-    protected void put(ColumnFamilyHandle cfHandle, WriteOptions writeOptions,
-        final byte[] keyBytes, final int keyLen,
-        final byte[] valueBytes, final int valueLen) throws RocksDBException {
-        if (!hold()) {
-            throw new IllegalStateException("rocksDB:" + this + " is not ready");
-        }
-        try {
-            this.db.put(cfHandle, writeOptions, keyBytes, 0, keyLen, valueBytes, 0, valueLen);
-        } catch (RocksDBException e) {
-            log.error("put Failed. {}, {}", this.dbPath, getStatusError(e));
-            throw e;
-        } finally {
-            release();
-        }
-    }
-
-    protected void put(ColumnFamilyHandle cfHandle, WriteOptions writeOptions,
-        final ByteBuffer keyBB, final ByteBuffer valueBB) throws RocksDBException {
-        if (!hold()) {
-            throw new IllegalStateException("rocksDB:" + this + " is not ready");
-        }
-        try {
-            this.db.put(cfHandle, writeOptions, keyBB, valueBB);
-        } catch (RocksDBException e) {
-            log.error("put Failed. {}, {}", this.dbPath, getStatusError(e));
-            throw e;
-        } finally {
-            release();
-        }
-    }
-
-    protected void batchPut(WriteOptions writeOptions, final WriteBatch batch) throws RocksDBException {
-        try {
-            this.db.write(writeOptions, batch);
-        } catch (RocksDBException e) {
-            log.error("batchPut Failed. {}, {}", this.dbPath, getStatusError(e));
-            throw e;
-        } finally {
-            batch.clear();
-        }
-    }
-
-    protected byte[] get(ColumnFamilyHandle cfHandle, ReadOptions readOptions, byte[] keyBytes) throws RocksDBException {
-        if (!hold()) {
-            throw new IllegalStateException("rocksDB:" + this + " is not ready");
-        }
-        try {
-            return this.db.get(cfHandle, readOptions, keyBytes);
-        } catch (RocksDBException e) {
-            log.error("get Failed. {}, {}", this.dbPath, getStatusError(e));
-            throw e;
-        } finally {
-            release();
-        }
-    }
-
-    protected int get(ColumnFamilyHandle cfHandle, ReadOptions readOptions, final ByteBuffer keyBB,
-        final ByteBuffer valueBB) throws RocksDBException {
-        if (!hold()) {
-            throw new IllegalStateException("rocksDB:" + this + " is not ready");
-        }
-        try {
-            return this.db.get(cfHandle, readOptions, keyBB, valueBB);
-        } catch (RocksDBException e) {
-            log.error("get Failed. {}, {}", this.dbPath, getStatusError(e));
-            throw e;
-        } finally {
-            release();
-        }
-    }
-
-    protected List<byte[]> multiGet(final ReadOptions readOptions,
-        final List<ColumnFamilyHandle> columnFamilyHandleList,
-        final List<byte[]> keys) throws RocksDBException {
-        if (!hold()) {
-            throw new IllegalStateException("rocksDB:" + this + " is not ready");
-        }
-        try {
-            return this.db.multiGetAsList(readOptions, columnFamilyHandleList, keys);
-        } catch (RocksDBException e) {
-            log.error("multiGet Failed. {}, {}", this.dbPath, getStatusError(e));
-            throw e;
-        } finally {
-            release();
-        }
-    }
-
-    protected void delete(ColumnFamilyHandle cfHandle, WriteOptions writeOptions,
-        byte[] keyBytes) throws RocksDBException {
-        if (!hold()) {
-            throw new IllegalStateException("rocksDB:" + this + " is not ready");
-        }
-        try {
-            this.db.delete(cfHandle, writeOptions, keyBytes);
-        } catch (RocksDBException e) {
-            log.error("delete Failed. {}, {}", this.dbPath, getStatusError(e));
-            throw e;
-        } finally {
-            release();
-        }
-    }
-
-    protected void delete(ColumnFamilyHandle cfHandle, WriteOptions writeOptions, ByteBuffer keyBB)
-        throws RocksDBException {
-        if (!hold()) {
-            throw new IllegalStateException("rocksDB:" + this + " is not ready");
-        }
-        try {
-            this.db.delete(cfHandle, writeOptions, keyBB);
-        } catch (RocksDBException e) {
-            log.error("delete Failed. {}, {}", this.dbPath, getStatusError(e));
-            throw e;
-        } finally {
-            release();
-        }
-    }
-
-    protected void rangeDelete(ColumnFamilyHandle cfHandle, WriteOptions writeOptions, final byte[] startKey,
-        final byte[] endKey) throws RocksDBException {
-        if (!hold()) {
-            throw new IllegalStateException("rocksDB:" + this + " is not ready");
-        }
-        try {
-            this.db.deleteRange(cfHandle, writeOptions, startKey, endKey);
-        } catch (RocksDBException e) {
-            log.error("rangeDelete Failed. {}, {}", this.dbPath, getStatusError(e));
-            throw e;
-        } finally {
-            release();
-        }
-    }
-
-    public void iterate(ColumnFamilyHandle columnFamilyHandle, final byte[] prefix, BiConsumer<byte[], byte[]> callback)
-        throws RocksDBException {
-
-        if (ArrayUtil.isEmpty(prefix)) {
-            throw new RocksDBException("Prefix is not allowed to be null");
-        }
-
-        iterate(columnFamilyHandle, prefix, null, null, callback);
-    }
-
-    public void iterate(ColumnFamilyHandle columnFamilyHandle, byte[] prefix,
-        final byte[] start, final byte[] end, BiConsumer<byte[], byte[]> callback) throws RocksDBException {
-
-        if (ArrayUtil.isEmpty(prefix) && ArrayUtil.isEmpty(start)) {
-            throw new RocksDBException("To determine lower boundary, prefix and start may not be null at the same "
-                + "time.");
-        }
-
-        if (ArrayUtil.isEmpty(prefix) && ArrayUtil.isEmpty(end)) {
-            throw new RocksDBException("To determine upper boundary, prefix and end may not be null at the same time.");
-        }
-
-        if (columnFamilyHandle == null) {
-            return;
-        }
-
-        ReadOptions readOptions = null;
-        Slice startSlice = null;
-        Slice endSlice = null;
-        Slice prefixSlice = null;
-        RocksIterator iterator = null;
-        try {
-            readOptions = new ReadOptions();
-            readOptions.setTotalOrderSeek(true);
-            readOptions.setReadaheadSize(4L * 1024 * 1024);
-            boolean hasStart = !ArrayUtil.isEmpty(start);
-            boolean hasPrefix = !ArrayUtil.isEmpty(prefix);
-
-            if (hasStart) {
-                startSlice = new Slice(start);
-                readOptions.setIterateLowerBound(startSlice);
-            }
-
-            if (!ArrayUtil.isEmpty(end)) {
-                endSlice = new Slice(end);
-                readOptions.setIterateUpperBound(endSlice);
-            }
-
-            if (!hasStart && hasPrefix) {
-                prefixSlice = new Slice(prefix);
-                readOptions.setIterateLowerBound(prefixSlice);
-            }
-
-            iterator = db.newIterator(columnFamilyHandle, readOptions);
-            if (hasStart) {
-                iterator.seek(start);
-            } else if (hasPrefix) {
-                iterator.seek(prefix);
-            }
-
-            while (iterator.isValid()) {
-                byte[] key = iterator.key();
-                if (hasPrefix && !checkPrefix(key, prefix)) {
-                    break;
-                }
-                callback.accept(iterator.key(), iterator.value());
-                iterator.next();
-            }
-        } finally {
-            if (startSlice != null) {
-                startSlice.close();
-            }
-            if (endSlice != null) {
-                endSlice.close();
-            }
-            if (prefixSlice != null) {
-                prefixSlice.close();
-            }
-            if (readOptions != null) {
-                readOptions.close();
-            }
-            if (iterator != null) {
-                iterator.close();
-            }
-        }
-    }
-
-    private boolean checkPrefix(byte[] key, byte[] upperBound) {
-        if (key.length < upperBound.length) {
-            return false;
-        }
-        for (int i = 0; i < upperBound.length; i++) {
-            if (key[i] > upperBound[i]) {
-                return false;
-            }
-        }
-        return true;
-    }
-
-
-
-    protected void open(final List<ColumnFamilyDescriptor> cfDescriptors) throws RocksDBException {
-        this.cfHandles.clear();
-        if (this.readOnly) {
-            this.db = RocksDB.openReadOnly(this.options, this.dbPath, cfDescriptors, cfHandles);
-        } else {
-            this.db = RocksDB.open(this.options, this.dbPath, cfDescriptors, cfHandles);
-        }
-        assert cfDescriptors.size() == cfHandles.size();
-
-        if (this.db == null) {
-            throw new RocksDBException("open rocksdb null");
-        }
-        try (Env env = this.db.getEnv()) {
-            env.setBackgroundThreads(8, Priority.LOW);
-        }
-    }
-
-    protected abstract boolean postLoad();
-
-    public synchronized boolean start() {
-        if (this.loaded) {
-            return true;
-        }
-        if (postLoad()) {
-            this.loaded = true;
-            log.info("RocksDB [{}] starts OK", this.dbPath);
-            this.closed = false;
-            return true;
-        } else {
-            return false;
-        }
-    }
-
-    /**
-     * Close column family handles except the default column family
-     */
-    protected abstract void preShutdown();
-
-    public synchronized boolean shutdown() {
-        try {
-            if (!this.loaded) {
-                log.info("RocksDBStorage is not loaded, shutdown OK. dbPath={}, readOnly={}", this.dbPath, this.readOnly);
-                return true;
-            }
-
-            final FlushOptions flushOptions = new FlushOptions();
-            flushOptions.setWaitForFlush(true);
-            try {
-                flush(flushOptions);
-            } catch (Throwable e) {
-                log.error("flush rocksdb wal failed when shutdown", e);
-            } finally {
-                flushOptions.close();
-            }
-            this.db.cancelAllBackgroundWork(true);
-            this.db.pauseBackgroundWork();
-            //The close order matters.
-            //1. close column family handles
-            preShutdown();
-
-            this.defaultCFHandle.close();
-
-            //2. close column family options.
-            for (final ColumnFamilyOptions opt : this.cfOptions) {
-                opt.close();
-            }
-            //3. close options
-            if (this.writeOptions != null) {
-                this.writeOptions.close();
-            }
-            //if (this.ableWalWriteOptions != null) {
-            //    this.ableWalWriteOptions.close();
-            //}
-            if (this.readOptions != null) {
-                this.readOptions.close();
-            }
-            if (this.totalOrderReadOptions != null) {
-                this.totalOrderReadOptions.close();
-            }
-            //4. close db.
-            if (db != null && !this.readOnly) {
-                try {
-                    this.db.syncWal();
-                } catch (Throwable e) {
-                    log.error("rocksdb sync wal failed when shutdown", e);
-                } finally {
-                    flushOptions.close();
-                }
-
-            }
-            if (db != null) {
-                try {
-                    this.db.closeE();
-                } catch (Throwable e) {
-                    log.error("rocksdb db closeE failed when shutdown", e);
-                }
-
-            }
-            // Close DBOptions after RocksDB instance is closed.
-            if (this.options != null) {
-                this.options.close();
-            }
-            //5. help gc.
-            this.cfOptions.clear();
-            this.db = null;
-            this.readOptions = null;
-            this.totalOrderReadOptions = null;
-            this.writeOptions = null;
-            //this.ableWalWriteOptions = null;
-            this.options = null;
-
-            this.loaded = false;
-            log.info("RocksDB shutdown OK. {}", this.dbPath);
-        } catch (Exception e) {
-            log.error("RocksDB shutdown failed. {}", this.dbPath, e);
-            return false;
-        }
-        return true;
-    }
-
-    public void flush(final FlushOptions flushOptions) throws RocksDBException {
-        flush(flushOptions, this.cfHandles);
-    }
-
-    public void flush(final FlushOptions flushOptions, List<ColumnFamilyHandle> columnFamilyHandles) throws RocksDBException {
-        if (!this.loaded || this.readOnly || closed) {
-            return;
-        }
-
-        try {
-            if (db != null) {
-                // For atomic-flush, we have to explicitly specify column family handles
-                // See https://github.com/rust-rocksdb/rust-rocksdb/pull/793
-                // and https://github.com/facebook/rocksdb/blob/8ad4c7efc48d301f5e85467105d7019a49984dc8/include/rocksdb/db.h#L1667
-                this.db.flush(flushOptions, columnFamilyHandles);
-            }
-        } catch (RocksDBException e) {
-            log.error("flush Failed. {}, {}", this.dbPath, getStatusError(e));
-            throw e;
-        }
-    }
-
-    public void flushWAL() throws RocksDBException {
-        this.db.flushWal(true);
-    }
-
-    public Statistics getStatistics() {
-        return this.options.statistics();
-    }
-
-    public ColumnFamilyHandle getDefaultCFHandle() {
-        return defaultCFHandle;
-    }
-
-    public List<LiveFileMetaData> getCompactionStatus() {
-        if (!hold()) {
-            return null;
-        }
-        try {
-            return this.db.getLiveFilesMetaData();
-        } finally {
-            release();
-        }
-    }
-
-
-    private String getStatusError(RocksDBException e) {
-        if (e == null || e.getStatus() == null) {
-            return "null";
-        }
-        Status status = e.getStatus();
-        StringBuilder sb = new StringBuilder(64);
-        sb.append("code: ");
-        if (status.getCode() != null) {
-            sb.append(status.getCode().name());
-        } else {
-            sb.append("null");
-        }
-        sb.append(", ").append("subCode: ");
-        if (status.getSubCode() != null) {
-            sb.append(status.getSubCode().name());
-        } else {
-            sb.append("null");
-        }
-        sb.append(", ").append("state: ").append(status.getState());
-        return sb.toString();
-    }
-
-    public void statRocksdb(Logger log) {
-        try {
-            // Log Memory Usage
-            String blockCacheMemUsage = this.db.getProperty("rocksdb.block-cache-usage");
-            String indexesAndFilterBlockMemUsage = this.db.getProperty("rocksdb.estimate-table-readers-mem");
-            String memTableMemUsage = this.db.getProperty("rocksdb.cur-size-all-mem-tables");
-            String blocksPinnedByIteratorMemUsage = this.db.getProperty("rocksdb.block-cache-pinned-usage");
-            log.info("RocksDB Memory Usage: BlockCache: {}, IndexesAndFilterBlock: {}, MemTable: {}, BlocksPinnedByIterator: {}",
-                blockCacheMemUsage, indexesAndFilterBlockMemUsage, memTableMemUsage, blocksPinnedByIteratorMemUsage);
-
-            // Log file metadata by level
-            List<LiveFileMetaData> liveFileMetaDataList = this.getCompactionStatus();
-            if (liveFileMetaDataList == null || liveFileMetaDataList.isEmpty()) {
-                return;
-            }
-            Map<Integer, StringBuilder> map = MapUtil.newHashMap();
-            for (LiveFileMetaData metaData : liveFileMetaDataList) {
-                StringBuilder sb = map.computeIfAbsent(metaData.level(), k -> new StringBuilder(256));
-                sb.append(new String(metaData.columnFamilyName(), StandardCharsets.UTF_8)).append(SPACE).
-                    append(metaData.fileName()).append(SPACE).
-                    append("file-size: ").append(metaData.size()).append(SPACE).
-                    append("number-of-entries: ").append(metaData.numEntries()).append(SPACE).
-                    append("file-read-times: ").append(metaData.numReadsSampled()).append(SPACE).
-                    append("deletions: ").append(metaData.numDeletions()).append(SPACE).
-                    append("being-compacted: ").append(metaData.beingCompacted()).append("\n");
-            }
-            map.forEach((key, value) -> log.info("level: {}\n{}", key, value.toString()));
-        } catch (Exception ignored) {
-        }
-    }
-}

+ 0 - 205
src/main/java/com/qingjian/common/rocksdb/DefaultRocksDBStorage.java

@@ -1,205 +0,0 @@
-package com.qingjian.common.rocksdb;
-
-import lombok.extern.slf4j.Slf4j;
-import org.rocksdb.*;
-import org.rocksdb.util.SizeUnit;
-
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.List;
-
-@Slf4j
-public class DefaultRocksDBStorage extends AbstractRocksDBStorage {
-
-    private static final byte[] IPS_DATA_COLUMN_FAMILY_NAME = "ispData".getBytes(StandardCharsets.UTF_8);
-    private static final byte[] CACHE_DATA_COLUMN_FAMILY_NAME = "cacheData".getBytes(StandardCharsets.UTF_8);
-
-    public ColumnFamilyHandle ipsDataFamilyHandle;
-    public ColumnFamilyHandle cacheDataFamilyHandle;
-
-    public DefaultRocksDBStorage(final String dbPath) {
-        this(dbPath, false);
-    }
-
-    public DefaultRocksDBStorage(final String dbPath, CompressionType compressionType) {
-        this(dbPath, false);
-        this.compressionType = compressionType;
-    }
-
-    public DefaultRocksDBStorage(final String dbPath, boolean readOnly) {
-        super(dbPath);
-        this.readOnly = readOnly;
-    }
-
-    protected void initOptions() {
-        this.options = createConfigDBOptions();
-        super.initOptions();
-    }
-
-    @Override
-    protected boolean postLoad() {
-        try {
-            initOptions();
-            final List<ColumnFamilyDescriptor> cfDescriptors = new ArrayList<>();
-            ColumnFamilyOptions defaultOptions = createConfigColumnFamilyOptions();
-            this.cfOptions.add(defaultOptions);
-            cfDescriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, defaultOptions));
-            cfDescriptors.add(new ColumnFamilyDescriptor(IPS_DATA_COLUMN_FAMILY_NAME, defaultOptions));
-            cfDescriptors.add(new ColumnFamilyDescriptor(CACHE_DATA_COLUMN_FAMILY_NAME, defaultOptions));
-            open(cfDescriptors);
-            //银行或者其他数据
-            this.defaultCFHandle = cfHandles.get(0);
-            //手机运营商数据
-            this.ipsDataFamilyHandle = cfHandles.get(1);
-            //Solon 持久化缓存数据
-            this.cacheDataFamilyHandle = cfHandles.get(2);
-
-        } catch (final Exception e) {
-            log.error("postLoad Failed. {}", this.dbPath, e);
-            return false;
-        }
-        return true;
-    }
-
-    @Override
-    protected void preShutdown() {
-        closeHandle(this.cacheDataFamilyHandle);
-        closeHandle(this.ipsDataFamilyHandle);
-    }
-
-    private void closeHandle(ColumnFamilyHandle handle) {
-        if (handle != null) {
-            handle.close();
-        }
-    }
-
-    public static ColumnFamilyOptions createConfigColumnFamilyOptions() {
-        BlockBasedTableConfig blockBasedTableConfig = new BlockBasedTableConfig().
-                setFormatVersion(5).
-                setIndexType(IndexType.kBinarySearch).
-                setDataBlockIndexType(DataBlockIndexType.kDataBlockBinarySearch).
-                setBlockSize(32 * SizeUnit.KB).
-                setFilterPolicy(new BloomFilter(16, false)).
-                setCacheIndexAndFilterBlocks(true).
-                setCacheIndexAndFilterBlocksWithHighPriority(true).
-                setPinL0FilterAndIndexBlocksInCache(false).
-                setPinTopLevelIndexAndFilter(true).
-                setBlockCache(new LRUCache(256 * SizeUnit.MB, 8, false)).
-                setWholeKeyFiltering(true);
-
-        ColumnFamilyOptions options = new ColumnFamilyOptions();
-        return options.setMaxWriteBufferNumber(4).
-                setWriteBufferSize(64 * SizeUnit.MB).
-                setMinWriteBufferNumberToMerge(1).
-                setTableFormatConfig(blockBasedTableConfig).
-                setMemTableConfig(new SkipListMemTableConfig()).
-                setCompressionType(CompressionType.LZ4_COMPRESSION).
-                setNumLevels(7).
-                setCompactionStyle(CompactionStyle.LEVEL).
-                setLevel0FileNumCompactionTrigger(4).
-                setLevel0SlowdownWritesTrigger(8).
-                setLevel0StopWritesTrigger(12).
-                setTargetFileSizeBase(64 * SizeUnit.MB).
-                setTargetFileSizeMultiplier(2).
-                setMaxBytesForLevelBase(256 * SizeUnit.MB).
-                setMaxBytesForLevelMultiplier(2).
-                setMergeOperator(new StringAppendOperator());
-    }
-
-    public static DBOptions createConfigDBOptions() {
-        DBOptions options = new DBOptions();
-        Statistics statistics = new Statistics();
-        statistics.setStatsLevel(StatsLevel.EXCEPT_DETAILED_TIMERS);
-        return options.
-                setInfoLogLevel(InfoLogLevel.INFO_LEVEL).
-                setWalRecoveryMode(WALRecoveryMode.SkipAnyCorruptedRecords).
-                setManualWalFlush(true).
-                setDbWriteBufferSize(128 * SizeUnit.MB).
-                setBytesPerSync(SizeUnit.MB).
-                setWalBytesPerSync(SizeUnit.MB).
-                setCreateIfMissing(true).
-                setCreateMissingColumnFamilies(true).
-                setMaxOpenFiles(-1).
-                setMaxLogFileSize(SizeUnit.GB).
-                setKeepLogFileNum(5).
-                setMaxManifestFileSize(SizeUnit.GB).
-                setAllowConcurrentMemtableWrite(true).
-                setStatistics(statistics).
-                setStatsDumpPeriodSec(600).
-                setMaxBackgroundJobs(32).
-                setMaxSubcompactions(4).
-                setDelayedWriteRate(16 * SizeUnit.MB);
-    }
-
-
-    public void put(final byte[] keyBytes, final int keyLen, final byte[] valueBytes) throws RocksDBException {
-        put(this.defaultCFHandle, this.writeOptions, keyBytes, keyLen, valueBytes, valueBytes.length);
-    }
-
-    public byte[] get(final byte[] keyBytes) throws RocksDBException {
-        return get(this.defaultCFHandle, this.totalOrderReadOptions, keyBytes);
-    }
-
-    public void updateIspData(final byte[] keyBytes, final byte[] valueBytes) throws RocksDBException {
-        put(this.ipsDataFamilyHandle, this.writeOptions, keyBytes, keyBytes.length, valueBytes, valueBytes.length);
-    }
-
-    public byte[] getIspData(final byte[] keyBytes) throws RocksDBException {
-        return get(this.ipsDataFamilyHandle, this.totalOrderReadOptions, keyBytes);
-    }
-
-    public void updateCacheData(final byte[] keyBytes, final byte[] valueBytes) throws RocksDBException {
-        put(this.cacheDataFamilyHandle, this.writeOptions, keyBytes, keyBytes.length, valueBytes, valueBytes.length);
-    }
-
-    public byte[] getCacheData(final byte[] keyBytes) throws RocksDBException {
-        return get(this.cacheDataFamilyHandle, this.totalOrderReadOptions, keyBytes);
-    }
-
-    public void deleteCacheData(final byte[] keyBytes) throws RocksDBException {
-        delete(this.cacheDataFamilyHandle, this.writeOptions, keyBytes);
-    }
-
-    public void delete(final byte[] keyBytes) throws RocksDBException {
-        delete(this.defaultCFHandle, this.writeOptions, keyBytes);
-    }
-
-    public List<byte[]> multiGet(final List<byte[]> keys) throws RocksDBException {
-        int size = keys.size();
-        List<ColumnFamilyHandle> cfhList = new ArrayList<>(size);
-        for (int i = 0; i < size; i++) {
-            cfhList.add(this.defaultCFHandle);
-        }
-        return multiGet(this.totalOrderReadOptions, cfhList, keys);
-    }
-
-    public List<byte[]> multiGet(ColumnFamilyHandle cfh, final List<byte[]> keys) throws RocksDBException {
-        int size = keys.size();
-        List<ColumnFamilyHandle> cfhList = new ArrayList<>(size);
-        for (int i = 0; i < size; i++) {
-            cfhList.add(cfh);
-        }
-        return multiGet(this.totalOrderReadOptions, cfhList, keys);
-    }
-
-    public List<byte[]> multiGet(final List<ColumnFamilyHandle> cfhList, final List<byte[]> keys) throws
-            RocksDBException {
-        return multiGet(this.totalOrderReadOptions, cfhList, keys);
-    }
-
-    public void batchPut(final WriteBatch batch) throws RocksDBException {
-        batchPut(this.writeOptions, batch);
-    }
-
-    public void batchIpsPut(final WriteBatch batch) throws RocksDBException {
-        batchPut(this.writeOptions, batch);
-    }
-
-    public RocksIterator iterator() {
-        return this.db.newIterator(this.defaultCFHandle, this.totalOrderReadOptions);
-    }
-
-    public RocksIterator iterator(ReadOptions readOptions) {
-        return this.db.newIterator(this.defaultCFHandle, readOptions);
-    }
-}

+ 0 - 173
src/main/java/com/qingjian/common/rocksdb/RocksDBHelper.java

@@ -1,173 +0,0 @@
-package com.qingjian.common.rocksdb;
-
-import cn.hutool.core.util.StrUtil;
-import lombok.extern.slf4j.Slf4j;
-import org.rocksdb.RocksDBException;
-import org.rocksdb.WriteBatch;
-import org.rocksdb.WriteOptions;
-
-import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * @author cc
- */
-@Slf4j
-public class RocksDBHelper {
-
-    public static final RocksDBHelper INSTANCE = new RocksDBHelper();
-
-    private static final String STORAGE_KEY = "storage_key";
-    private final Map<String, DefaultRocksDBStorage> storageMap = new HashMap<>(1);
-
-    private RocksDBHelper() {
-
-    }
-
-    public void put(DefaultRocksDBStorage storage) {
-        storageMap.put(STORAGE_KEY, storage);
-    }
-
-    public DefaultRocksDBStorage get() {
-        return storageMap.get(STORAGE_KEY);
-    }
-
-    // JVM 关闭时释放资源
-    public void remove() {
-        DefaultRocksDBStorage oldStorage = storageMap.remove(STORAGE_KEY);
-        if (oldStorage != null) {
-            oldStorage.shutdown();
-        }
-    }
-
-    public void put(final byte[] key, final byte[] value) throws RocksDBException {
-        DefaultRocksDBStorage storage = get();
-        if (storage == null) {
-            return;
-        }
-        storage.put(key, key.length, value);
-    }
-
-    public byte[] get(final byte[] key) throws RocksDBException {
-        DefaultRocksDBStorage storage = get();
-        if (storage == null) {
-            return null;
-        }
-        return storage.get(key);
-    }
-
-    public String get(final String key) {
-        try {
-            byte[] bytes = get(StrUtil.utf8Bytes(key));
-            if (bytes != null) {
-                return StrUtil.utf8Str(bytes);
-            }
-        } catch (Exception ignored) {
-        }
-        return null;
-    }
-
-    public String getIspValue(String key) {
-        try {
-            byte[] bytes = get().getIspData(StrUtil.utf8Bytes(key));
-            if (bytes != null) {
-                return StrUtil.utf8Str(bytes);
-            }
-        } catch (Exception ignored) {
-        }
-        return "未知运营商";
-    }
-
-    public Map<String, String> batchGetIspValue(List<String> keys) {
-        DefaultRocksDBStorage storage = get();
-        if (storage == null) {
-            return Map.of();
-        }
-        List<byte[]> keyBytes = keys.stream().map(StrUtil::utf8Bytes).toList();
-        try {
-            List<byte[]> results = storage.multiGet(storage.ipsDataFamilyHandle, keyBytes);
-            Map<String, String> map = HashMap.newHashMap(keys.size());
-            for (int i = 0; i < keys.size(); i++) {
-                byte[] value = results.get(i);
-                if (value != null) {
-                    map.put(keys.get(i), new String(value, StandardCharsets.UTF_8));
-                }
-            }
-            return map;
-        } catch (RocksDBException e) {
-            log.error("batchGetIspValue failed", e);
-            return Map.of();
-        }
-    }
-
-    public Map<String, String> batchGet(List<String> keys) {
-        DefaultRocksDBStorage storage = get();
-        if (storage == null) {
-            return Map.of();
-        }
-        List<byte[]> keyBytes = keys.stream().map(StrUtil::utf8Bytes).toList();
-        try {
-            List<byte[]> results = storage.multiGet(keyBytes);
-            Map<String, String> map = HashMap.newHashMap(keys.size());
-            for (int i = 0; i < keys.size(); i++) {
-                byte[] value = results.get(i);
-                if (value != null) {
-                    map.put(keys.get(i), new String(value, StandardCharsets.UTF_8));
-                }
-            }
-            return map;
-        } catch (RocksDBException e) {
-            log.error("batchGet failed", e);
-            return Map.of();
-        }
-    }
-
-    /**
-     * 增(批量)
-     */
-    public void batchPut(Map<String, String> map) throws RocksDBException {
-        DefaultRocksDBStorage storage = get();
-        if (storage == null) {
-            return;
-        }
-        try (WriteBatch writeBatch = new WriteBatch()) {
-            WriteOptions writeOptions = new WriteOptions();
-            for (Map.Entry<String, String> entry : map.entrySet()) {
-                String key = entry.getKey();
-                String value = entry.getValue();
-                writeBatch.put(StrUtil.utf8Bytes(key), StrUtil.utf8Bytes(value));
-            }
-            storage.batchPut(writeOptions, writeBatch);
-        }
-    }
-
-    public void batchIpsPut(Map<String, String> map) throws RocksDBException {
-        DefaultRocksDBStorage storage = get();
-        if (storage == null) {
-            return;
-        }
-        try (WriteBatch writeBatch = new WriteBatch()) {
-            WriteOptions writeOptions = new WriteOptions();
-            for (Map.Entry<String, String> entry : map.entrySet()) {
-                String key = entry.getKey();
-                String value = entry.getValue();
-                writeBatch.put(storage.ipsDataFamilyHandle, StrUtil.utf8Bytes(key), StrUtil.utf8Bytes(value));
-            }
-            storage.batchPut(writeOptions, writeBatch);
-        }
-    }
-
-    /**
-     * 删
-     */
-    public void delete(String key) throws RocksDBException {
-        DefaultRocksDBStorage storage = get();
-        if (storage == null) {
-            return;
-        }
-        storage.delete(StrUtil.utf8Bytes(key));
-    }
-
-}

+ 5 - 5
src/main/java/com/qingjian/common/utils/DataUtil.java

@@ -1,8 +1,8 @@
 package com.qingjian.common.utils;
 
+import cn.hutool.core.convert.Convert;
 import cn.hutool.core.util.NumberUtil;
 import cn.hutool.core.util.StrUtil;
-import com.qingjian.common.constants.PathConst;
 
 import java.math.BigDecimal;
 
@@ -21,9 +21,9 @@ public class DataUtil {
      * @param cardNo 银行卡号
      * @return 格式化的银行卡 BIN 码(带前缀)
      */
-    public static String getCardPrefix6(String cardNo) {
+    public static int getCardPrefix6(String cardNo) {
         String prefix = StrUtil.sub(cardNo, 0, 6);
-        return PathConst.BANK_PREFIX + prefix;
+        return Convert.toInt(prefix, 0);
     }
 
     /**
@@ -33,9 +33,9 @@ public class DataUtil {
      * @param phone 手机号码
      * @return 格式化的手机号段(带前缀)
      */
-    public static String getPhonePrefix7(String phone) {
+    public static int getPhonePrefix7(String phone) {
         String prefix = StrUtil.sub(phone, 0, 7);
-        return PathConst.PHONE_PREFIX + prefix;
+        return Convert.toInt(prefix, 0);
     }
 
     /**

+ 10 - 18
src/main/java/com/qingjian/module/dm/clean/AbstractCallRecordDataLoader.java

@@ -1,18 +1,14 @@
 package com.qingjian.module.dm.clean;
 
 import cn.hutool.core.text.StrPool;
-import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.core.util.StrUtil;
-import cn.hutool.crypto.digest.DigestUtil;
 import com.qingjian.common.cache.CaseDataCache;
+import com.qingjian.common.cache.RogueBaseDataCache;
 import com.qingjian.common.model.call.entity.CallRecord;
 import com.qingjian.common.model.otg.dto.SearchDTO;
-import com.qingjian.common.rocksdb.RocksDBHelper;
 import com.qingjian.common.utils.DataUtil;
 import com.qingjian.common.utils.LuceneManager;
-import com.qingjian.common.utils.TowerUtils;
 import lombok.extern.slf4j.Slf4j;
-import org.rocksdb.RocksDBException;
 
 import java.sql.SQLException;
 
@@ -30,15 +26,11 @@ public abstract class AbstractCallRecordDataLoader<S> extends AbstractDataLoader
         String otherName = record.getOtherName();
         String personPhone = record.getPersonPhone();
         String otherPhone = record.getOtherPhone();
+        int phonePrefix7 = DataUtil.getPhonePrefix7(personPhone);
         if (StrUtil.isEmpty(operatorId) && StrUtil.isNotEmpty(personPhone)) {
-            record.setOperatorId(RocksDBHelper.INSTANCE.getIspValue(DataUtil.getPhonePrefix7(personPhone)));
+            record.setOperatorId(RogueBaseDataCache.INSTANCE.get(phonePrefix7));
         } else if (StrUtil.isNotEmpty(operatorId) && StrUtil.isNotEmpty(personPhone)) {
-            byte[] key = StrUtil.utf8Bytes(personPhone);
-            try {
-                RocksDBHelper.INSTANCE.put(key, StrUtil.utf8Bytes(operatorId));
-            } catch (RocksDBException e) {
-                log.error("[数据入库]RocksDB新增运营上和电话关联关系异常", e);
-            }
+            RogueBaseDataCache.INSTANCE.put(phonePrefix7, operatorId);
         }
         if (StrUtil.isEmpty(personName) && StrUtil.isNotEmpty(personPhone)) {
             //先从本文档获取
@@ -47,20 +39,20 @@ public abstract class AbstractCallRecordDataLoader<S> extends AbstractDataLoader
             //    //再从全局人员库获取
             //    filePersonName = CaseDataCache.findPersonVal(personPhone);
             //}
-            record.setPersonName(CaseDataCache.findPersonVal(personPhone));
+            record.setPersonName(CaseDataCache.PERSON_PHONE.get(personPhone));
         } else if (StrUtil.isNotEmpty(personName) && StrUtil.isNotEmpty(personPhone)) {
             CaseDataCache.putPerLib(personName + StrPool.UNDERLINE + 2, personPhone, record.getOperatorId());
             //CleanCache.putPerLib(personName + StrPool.UNDERLINE + record.getFileId(), personPhone);
         }
 
         if (StrUtil.isEmpty(otherName) && StrUtil.isNotEmpty(otherPhone)) {
-            record.setOtherName(CaseDataCache.findPersonVal(otherPhone));
+            record.setOtherName(CaseDataCache.PERSON_PHONE.get(otherPhone));
         } else if (StrUtil.isNotEmpty(otherName) && StrUtil.isNotEmpty(otherPhone)) {
             CaseDataCache.putPerLib(otherName + StrPool.UNDERLINE + 2, otherPhone, null);
         }
         String lac = record.getPersonRegionId();
         String cellId = record.getPersonCellId();
-        if (StrUtil.isNotEmpty(lac) && StrUtil.isNotEmpty(cellId)) {
+       /* if (StrUtil.isNotEmpty(lac) && StrUtil.isNotEmpty(cellId)) {
             Integer mnc = phone(record.getOperatorId());
             if (mnc != -1) {
                 try {
@@ -89,9 +81,9 @@ public abstract class AbstractCallRecordDataLoader<S> extends AbstractDataLoader
                     log.error("基站无法解析 lac={},cellId={}", lac, cellId);
                 }
             }
-        }
-        String personLibId = DigestUtil.md5Hex(personName + personPhone + "2");
-        record.setPersonLibId(personLibId);
+        }*/
+        //String personLibId = DigestUtil.md5Hex(personName + personPhone + "2");
+        //record.setPersonLibId(personLibId);
         CleanCache.addPerson(personName);
     }
 

+ 3 - 1
src/main/java/com/qingjian/module/dm/clean/AbstractDataLoader.java

@@ -36,8 +36,10 @@ public abstract class AbstractDataLoader<S, T extends Cleaned> implements DataLo
             return;
         }
         T t = convert(record);
+        long l = System.currentTimeMillis();
         fillData(t);
-        doWrite(t);
+        log.info("写入 fillData 成功, tableName={}, cost={}", tableName, System.currentTimeMillis() - l);
+        //doWrite(t);
     }
 
     @Override

+ 8 - 14
src/main/java/com/qingjian/module/dm/clean/AbstractTransRecordDataLoader.java

@@ -2,16 +2,14 @@ package com.qingjian.module.dm.clean;
 
 import cn.hutool.core.text.StrPool;
 import cn.hutool.core.util.StrUtil;
-import cn.hutool.crypto.digest.DigestUtil;
 import com.google.common.base.Joiner;
 import com.qingjian.common.cache.CaseDataCache;
+import com.qingjian.common.cache.RogueBaseDataCache;
 import com.qingjian.common.model.otg.dto.SearchDTO;
 import com.qingjian.common.model.trans.entity.TransRecord;
-import com.qingjian.common.rocksdb.RocksDBHelper;
 import com.qingjian.common.utils.DataUtil;
 import com.qingjian.common.utils.LuceneManager;
 import lombok.extern.slf4j.Slf4j;
-import org.rocksdb.RocksDBException;
 
 import java.sql.SQLException;
 
@@ -29,28 +27,24 @@ public abstract class AbstractTransRecordDataLoader<S> extends AbstractDataLoade
         String otherName = record.getOtherName();
         String personCardNo = record.getPersonCardNo();
         String otherCardNo = record.getOtherCardNo();
+        int cardPrefix6 = DataUtil.getCardPrefix6(personCardNo);
         if (StrUtil.isEmpty(bankName) && StrUtil.isNotEmpty(personCardNo)) {
-            record.setBankName(RocksDBHelper.INSTANCE.get(DataUtil.getCardPrefix6(personCardNo)));
+            record.setBankName(RogueBaseDataCache.INSTANCE.get(cardPrefix6));
         } else if (StrUtil.isNotEmpty(bankName) && StrUtil.isNotEmpty(personCardNo)) {
-            byte[] key = StrUtil.utf8Bytes(personCardNo);
-            try {
-                RocksDBHelper.INSTANCE.put(key, StrUtil.utf8Bytes(bankName));
-            } catch (RocksDBException e) {
-                log.error(e.getMessage(), e);
-            }
+            RogueBaseDataCache.INSTANCE.put(cardPrefix6, bankName);
         }
         if (StrUtil.isEmpty(personName) && StrUtil.isNotEmpty(personCardNo)) {
-            record.setPersonName(CaseDataCache.findPersonVal(personCardNo));
+            record.setPersonName(CaseDataCache.PERSON_CARD.get(personCardNo));
         } else if (StrUtil.isNotEmpty(personName) && StrUtil.isNotEmpty(personCardNo)) {
             CaseDataCache.putPerLib(personName + StrPool.UNDERLINE + 3, personCardNo, record.getBankName());
         }
         if (StrUtil.isEmpty(otherName) && StrUtil.isNotEmpty(otherCardNo)) {
-            record.setOtherName(CaseDataCache.findPersonVal(otherCardNo));
+            record.setOtherName(CaseDataCache.PERSON_CARD.get(otherCardNo));
         } else if (StrUtil.isNotEmpty(otherName) && StrUtil.isNotEmpty(otherCardNo)) {
             CaseDataCache.putPerLib(otherName + StrPool.UNDERLINE + 3, otherCardNo, null);
         }
-        String personLibId = DigestUtil.md5Hex(personName + personCardNo + "3");
-        record.setPersonLibId(personLibId);
+        //String personLibId = DigestUtil.md5Hex(personName + personCardNo + "3");
+        //record.setPersonLibId(personLibId);
 
         //扩展字段1
         String transType = record.getTransType();

+ 1 - 3
src/main/java/com/qingjian/module/person/controller/DataProfileController.java

@@ -2,7 +2,6 @@ package com.qingjian.module.person.controller;
 
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.qingjian.common.base.Query;
-import com.qingjian.common.constants.StrConsts;
 import com.qingjian.common.model.call.entity.CallOpenInfo;
 import com.qingjian.common.model.person.dto.*;
 import com.qingjian.common.model.person.query.*;
@@ -12,7 +11,6 @@ import com.qingjian.common.model.trans.query.TransBankCardInfoQuery;
 import com.qingjian.module.person.service.DataProfileService;
 import com.qingjian.module.person.service.IntimacyService;
 import org.noear.solon.annotation.*;
-import org.noear.solon.data.annotation.Cache;
 
 import java.util.List;
 
@@ -195,7 +193,7 @@ public class DataProfileController {
     //"亲密度-汇总")
     @Get
     @Mapping("/intimacySummary")
-    @Cache(key = "dp_intimacySummary_${personName}", tags = StrConsts.CACHE_TAG_QING_JIAN)
+    //@Cache(key = "dp_intimacySummary_${personName}", tags = StrConsts.CACHE_TAG_QING_JIAN)
     public List<IntimacyDTO> intimacySummary(String personName) {
         return intimacyService.intimacySummary(personName);
     }

+ 14 - 0
src/main/java/com/qingjian/module/plat/service/SseService.java

@@ -7,6 +7,7 @@ import com.qingjian.common.utils.Json;
 import lombok.extern.slf4j.Slf4j;
 import org.noear.solon.Utils;
 import org.noear.solon.annotation.Managed;
+import org.noear.solon.scheduling.annotation.Scheduled;
 import org.noear.solon.web.sse.SseEmitter;
 import org.noear.solon.web.sse.SseEvent;
 
@@ -58,4 +59,17 @@ public class SseService {
             log.error("sse心跳失败", e);
         }
     }
+
+    @Scheduled(fixedRate = 5000)
+    public void fixedSee() {
+        if (sseEmitter == null) {
+            return;
+        }
+        try {
+            String json = Json.toStr(new SseDTO(999, -1, "心跳", new SeeDataDTO(0, 0)));
+            sseEmitter.send(new SseEvent().id(Utils.guid()).data(json).reconnectTime(5000L));
+        } catch (Exception e) {
+            log.error("sse心跳失败", e);
+        }
+    }
 }

+ 2 - 0
src/test/java/com/qingjian/common/cache/RocksDBCacheTest.java

@@ -1,3 +1,4 @@
+/*
 package com.qingjian.common.cache;
 
 import com.fasterxml.jackson.core.type.TypeReference;
@@ -93,3 +94,4 @@ public class RocksDBCacheTest {
         }
     }
 }
+*/