import org.rocksdb.FlushOptions; //导入依赖的package包/类
/**
 * Flush the data in memtable of RocksDB into disk, and then create checkpoint
 * 
 * @param batchId
 */
@Override
public void checkpoint(long batchId) {
    long startTime = System.currentTimeMillis();
    try {
        rocksDb.flush(new FlushOptions());
        Checkpoint cp = Checkpoint.create(rocksDb);
        cp.createCheckpoint(getLocalCheckpointPath(batchId));
    } catch (RocksDBException e) {
        LOG.error("Failed to create checkpoint for batch-" + batchId, e);
        throw new RuntimeException(e.getMessage());
    }

    if (JStormMetrics.enabled)
        rocksDbFlushAndCpLatency.update(System.currentTimeMillis() - startTime);
}
 

import org.rocksdb.FlushOptions; //导入依赖的package包/类
@Override
protected void dbShutdownHandler()
  throws Exception
{
  log.alarm("RocksDB: flushing");
  FlushOptions fl = new FlushOptions();
  fl.setWaitForFlush(true);
  db.flush(fl);
  log.alarm("RocksDB: flush complete");


}
 

import org.rocksdb.FlushOptions; //导入依赖的package包/类
private static void rocksDbTest(RocksDB db, List<ColumnFamilyHandle> handlers) {
    try {
        ColumnFamilyHandle handler1 = null;
        ColumnFamilyHandle handler2 = null;
        if (handlers.size() > 0) {
            // skip default column family
            handler1 = handlers.get(1);
            handler2 = handlers.get(2);
        } else {
            handler1 = db.createColumnFamily(new ColumnFamilyDescriptor("test1".getBytes()));
            handler2 = db.createColumnFamily(new ColumnFamilyDescriptor("test2".getBytes()));
        }
        int startValue1 = getStartValue(db, handler1);
        int startValue2 = getStartValue(db, handler2);;

        Checkpoint cp = Checkpoint.create(db);
   
        if (isCompaction) {
            db.compactRange();
            LOG.info("Compaction!");
        }

        long flushWaitTime = System.currentTimeMillis() + flushInterval;
        for (int i = 0; i < putNum || putNum == -1; i++) {
            db.put(handler1, String.valueOf(i % 1000).getBytes(), String.valueOf(startValue1 + i).getBytes());
            db.put(handler2, String.valueOf(i % 1000).getBytes(), String.valueOf(startValue2 + i).getBytes());
            if (isFlush && flushWaitTime <= System.currentTimeMillis()) {
                db.flush(new FlushOptions());
                if (isCheckpoint) {
                    cp.createCheckpoint(cpPath + "/" + i);
                }
                flushWaitTime = System.currentTimeMillis() + flushInterval;
            }
        }
    } catch (RocksDBException e) {
        LOG.error("Failed to put or flush", e);
    }
}
 

import org.rocksdb.FlushOptions; //导入依赖的package包/类
FeatureStoreRocksDb(MetricsContext metricsContext, File dbPath) {

    MetricRegistry metrics = metricsContext.metrics();
    String context = metricsContext.context();

    putTimer = metrics.timer(MetricRegistry.name(
        context + "." + METRICS_PATH, "putTimer"));
    putMeter = metrics.meter(MetricRegistry.name(
        context + "." + METRICS_PATH, "putMeter"));
    this.loadAllTimer = metrics.timer(MetricRegistry.name(
        context + "." + METRICS_PATH, "loadAllTimer"));
    this.loadAllMeter = metrics.meter(MetricRegistry.name(
        context + "." + METRICS_PATH, "loadAllMeter"));

    this.findAllTimer = metrics.timer(MetricRegistry.name(
        context + "." + METRICS_PATH, "findAllTimer"));
    this.findAllMeter = metrics.meter(MetricRegistry.name(
        context + "." + METRICS_PATH, "findAllMeter"));

    BlockBasedTableConfig tableConfig = new BlockBasedTableConfig();
    tableConfig.setBlockCacheSize(BLOCK_CACHE_SIZE);
    tableConfig.setBlockSize(BLOCK_SIZE);

    options = new Options();
    options.setTableFormatConfig(tableConfig);
    options.setWriteBufferSize(WRITE_BUFFER_SIZE);
    options.setCompressionType(COMPRESSION_TYPE);
    options.setCompactionStyle(COMPACTION_STYLE);
    options.setMaxWriteBufferNumber(MAX_WRITE_BUFFER_NUMBER);
    options.setCreateIfMissing(CREATE_IF_MISSING);
    options.setErrorIfExists(ERROR_IF_EXISTS);
    writeOptions = new WriteOptions();
    writeOptions.setDisableWAL(DISABLE_WAL);
    writeOptions.setSync(true);
    readOptions = new ReadOptions();
    readOptions.setVerifyChecksums(true);
    readOptions.setFillCache(true);
    flushOptions = new FlushOptions();
    flushOptions.setWaitForFlush(WAIT_FOR_FLUSH);

    final File parent = new File("/tmp", "outland");
    rocksDir = new File(dbPath, "feature-store");
    //noinspection ResultOfMethodCallIgnored
    rocksDir.getParentFile().mkdirs();
    rocks = initializeRocksDb(); // todo: move this out?
  }
 

import org.rocksdb.FlushOptions; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public void openDB(ProcessorContext context) {
    // initialize the default rocksdb options
    final BlockBasedTableConfig tableConfig = new BlockBasedTableConfig();
    tableConfig.setBlockCacheSize(BLOCK_CACHE_SIZE);
    tableConfig.setBlockSize(BLOCK_SIZE);

    options = new Options();
    options.setTableFormatConfig(tableConfig);
    options.setWriteBufferSize(WRITE_BUFFER_SIZE);
    options.setCompressionType(COMPRESSION_TYPE);
    options.setCompactionStyle(COMPACTION_STYLE);
    options.setMaxWriteBufferNumber(MAX_WRITE_BUFFERS);
    options.setCreateIfMissing(true);
    options.setErrorIfExists(false);
    options.setInfoLogLevel(InfoLogLevel.ERROR_LEVEL);
    // this is the recommended way to increase parallelism in RocksDb
    // note that the current implementation of setIncreaseParallelism affects the number
    // of compaction threads but not flush threads (the latter remains one). Also
    // the parallelism value needs to be at least two because of the code in
    // https://github.com/facebook/rocksdb/blob/62ad0a9b19f0be4cefa70b6b32876e764b7f3c11/util/options.cc#L580
    // subtracts one from the value passed to determine the number of compaction threads
    // (this could be a bug in the RocksDB code and their devs have been contacted).
    options.setIncreaseParallelism(Math.max(Runtime.getRuntime().availableProcessors(), 2));

    wOptions = new WriteOptions();
    wOptions.setDisableWAL(true);

    fOptions = new FlushOptions();
    fOptions.setWaitForFlush(true);

    final Map<String, Object> configs = context.appConfigs();
    final Object configSetterValue = configs.get(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG);
    final Class<RocksDBConfigSetter> configSetterClass = (Class<RocksDBConfigSetter>) ConfigDef.parseType(
            StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG,
            configSetterValue,
            ConfigDef.Type.CLASS);

    if (configSetterClass != null) {
        final RocksDBConfigSetter configSetter = Utils.newInstance(configSetterClass);
        configSetter.setConfig(name, options, configs);
    }
    // we need to construct the serde while opening DB since
    // it is also triggered by windowed DB segments without initialization
    this.serdes = new StateSerdes<>(
        ProcessorStateManager.storeChangelogTopic(context.applicationId(), name),
        keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
        valueSerde == null ? (Serde<V>) context.valueSerde() : valueSerde);

    this.dbDir = new File(new File(context.stateDir(), parentDir), this.name);
    try {
        this.db = openDB(this.dbDir, this.options, TTL_SECONDS);
    } catch (IOException e) {
        throw new StreamsException(e);
    }
}