RocksDB应用原创

# 1. 添加依赖

    <properties>
        <rocksdbjni.version>6.14.6</rocksdbjni.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.rocksdb</groupId>
            <artifactId>rocksdbjni</artifactId>
            <version>${rocksdbjni.version}</version>
        </dependency>
    </dependencies>

1
2
3
4
5
6
7
8
9
10

# 2. Factory工厂类

import org.apache.commons.io.FileUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.rocksdb.Options;
import org.rocksdb.RocksDB;

import java.io.File;

/**
 * @Auther:BigData-weic
 * @ClassName:RocksDBFactory
 * @Date:2021/8/6 15:05
 * @功能描述: RocksDB工厂类
 * @Version:1.0
 */
public class RocksDBFactory {
    private static final Logger logger = LogManager.getLogger("RocksDBFactory");

    /* *
     * 功能描述：打开数据库
     *
     * @param: [dbpath 数据库初始化位置, dateStr 时间做区分, keySpace 数据库名字]
     * @return: org.rocksdb.RocksDB
     */
    public static RocksDB createRocksDB(String dbpath, String dateStr, String keySpace) throws Exception {
        String dbPath = dbpath + File.separator + dateStr + File.separator + keySpace;
        Options options = new Options();
        options.setCreateIfMissing(true);
        FileUtils.forceMkdir(new File(dbPath));
        RocksDB rocksDB = RocksDB.open(options, dbPath);
        logger.info("Init rocksdb dbpath: {}", dbPath);
        return rocksDB;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# 3. 多线程初始化

//从配置文件中获取分片个数，也就是数据库个数
private int shardCount = ConfigurationUtils.getConfig().getInt("rocksdb.shard-count", 100);
private RocksDB[] rocksDBs = new RocksDB[shardCount];

public ExtractorClass(String dataStr) throws Exception {
    // 加载类时，初始化rocksDB数据库
    ExecutorService threadPoolExecutor = null;
    ArrayList<Future> futures = new ArrayList<>();

    try {
        threadPoolExecutor = Executors.newFixedThreadPool(20);
        for (int i = 0; i < shardCount; i++) {
            int index = i;
            Future<Boolean> future = threadPoolExecutor.submit(() -> {
                try {
                    rocksDBs[index]  = RocksDBFactory.createRocksDB(ConfigurationUtils.getConfig().getString("rocksdb.dbpath"), dataStr, ConfigurationUtils.getConfig().getString("rocksdb.keyspace") + index);
                    return true;
                } catch (Exception e) {
                    logger.error(e.getMessage(),e);
                    return false;
                }
            });
            futures.add(future);
        }

        for (Future<Boolean> future : futures) {
            if (!future.get()) {
                throw new Exception("RocksDB init exception");
            }
        }
    } finally {
        if (threadPoolExecutor != null) {
            threadPoolExecutor.shutdown();
        }
    }
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# 4. 关闭RocksDB资源

public void closeRocksDBs() {
    for (int i = 0; i < rocksDBs.length; i++) {
        rocksDBs[i].close();
    }
}

1
2
3
4
5

# 5. 初始化单个数据库

private RocksDB rocksDB;
//初始化数据库地址从配置文件加载
public ExtractorClass(String dataStr) throws Exception {
    Configuration config = ConfigurationUtils.getConfig();
    rocksDB = RocksDBFactory.createRocksDB(config.getString("rocksdb.dbpath"), dataStr, config.getString("rocksdb.keyspace"));
}

1
2
3
4
5
6

# 6. 数据写入和查询

该操作可做数据去重使用

String record = "hehe";
int index = (int) (Math.abs(hash.hash(record)) % shardCount);
byte[] value = rocksDBs[index].get(record.getBytes(StandardCharsets.UTF_8));
if (value == null) {
    rocksDBs[index].put(record.getBytes(StandardCharsets.UTF_8), "".getBytes(StandardCharsets.UTF_8));
    //数据写出write()
}

1
2
3
4
5
6
7

← Java加密解密 Java IO读写文件→