3、RocksDB 详解：RocksDB 是什么、RocksDB 在 Kafka Streams 中的角色、RocksDB 配置与调优

好，咱们进入状态存储的核心——RocksDB。说实话，Kafka Streams 的持久化能力，大半都靠它撑着。你想想看，一个流处理应用跑上几天几周，中间状态要是全放内存，那内存得爆成什么样？RocksDB 就是来解决这个问题的。

3.1 RocksDB 是什么

RocksDB 是一个嵌入式、持久化的键值存储引擎。它由 Facebook 在 2013 年开源，基于 Google 的 LevelDB 改造而来。说白了，它就是一个跑在你应用进程里的数据库，不需要单独部署服务。

它的核心特性我列一下：

• LSM-Tree 结构：写操作先写内存（MemTable），满了再刷到磁盘（SSTable）。写入吞吐极高。
• 列族（Column Family）：支持逻辑分区，Kafka Streams 的每个状态存储就是一个列族。
• 压缩与合并：后台自动做 compaction，清理过期数据，减少读放大。
• 块缓存（Block Cache）：热数据留在内存，冷数据从磁盘读，兼顾性能和容量。

我个人习惯把 RocksDB 比作「带持久化的 ConcurrentHashMap」。你 get 和 put 的体验跟内存 Map 差不多，但数据不会丢。嗯，这里要注意：它不是关系型数据库，不支持 SQL，只做键值操作。

3.2 RocksDB 在 Kafka Streams 中的角色

在 Kafka Streams 里，RocksDB 扮演什么角色？我直接说结论：它是状态存储的「物理载体」。

每次你调用 KStream#groupByKey().aggregate() 或者 KStream#join()，Kafka Streams 都会在本地创建一个 RocksDB 实例来保存中间结果。这个实例对应一个状态存储（State Store），而状态存储又对应一个 Kafka 内部的 changelog topic。

整个链路是这样的：

1. 应用写入状态 → 写入 RocksDB（本地磁盘）
2. 同时写入 changelog topic（Kafka 集群）
3. 宕机恢复时 → 从 changelog topic 回放数据到 RocksDB

你看，RocksDB 负责「本地快速读写」，Kafka 负责「远程容灾备份」。两者配合，既保证了性能，又保证了数据不丢。

我曾经遇到一个生产问题：某个聚合任务的状态存储越来越大，最后磁盘满了。排查下来发现是 RocksDB 的 compaction 配置没调好，历史数据没及时清理。从那以后，我对 RocksDB 的配置就格外上心。

下面这张图展示了 Kafka Streams 与 RocksDB 的关系：

3.3 RocksDB 配置与调优

RocksDB 的配置项非常多，但咱们不需要全记住。我挑几个在 Kafka Streams 场景下最关键的来说。

3.3.1 内存相关配置

内存是 RocksDB 性能的关键。配置不合理，要么 OOM，要么性能拉胯。

配置项	默认值	说明
write_buffer_size	64 MB	单个 MemTable 的大小。写多读少的场景可以调大。
max_write_buffer_number	2	最多同时存在的 MemTable 数量。调大可以缓解写压力。
block_cache_size	8 MB	块缓存大小。读密集场景建议调大。
total_memory	无	RocksDB 总内存上限。建议设置为 JVM 堆外内存的 60%-70%。

3.3.2 Compaction 相关配置

Compaction 是 RocksDB 的后台合并操作。配置不好，会导致写放大严重，或者磁盘空间暴涨。

配置项	默认值	说明
compaction_style	kCompactionStyleLevel	层级合并，默认就是最优选择。
max_background_compactions	1	后台 compaction 线程数。多核机器可以调大。
target_file_size_base	64 MB	L1 层 SSTable 的目标大小。调大可以减少文件数。

3.3.3 在 Kafka Streams 中配置 RocksDB

在 Kafka Streams 里配置 RocksDB，有两种方式：

方式一：通过 Properties 全局配置

Properties props = new Properties();
props.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, 
          CustomRocksDBConfig.class.getName());

方式二：实现 RocksDBConfigSetter 接口

public class CustomRocksDBConfig implements RocksDBConfigSetter {
    @Override
    public void setConfig(final String storeName, 
                          final Options options, 
                          final Map<String, Object> configs) {
        // 设置 MemTable 大小
        options.setWriteBufferSize(128 * 1024 * 1024L); // 128 MB
        
        // 设置块缓存
        final BlockBasedTableConfig tableConfig = new BlockBasedTableConfig();
        tableConfig.setBlockCacheSize(256 * 1024 * 1024L); // 256 MB
        options.setTableFormatConfig(tableConfig);
        
        // 设置 compaction 线程数
        options.setMaxBackgroundCompactions(2);
    }
    
    @Override
    public void close(final String storeName, final Options options) {
        // 清理资源
    }
}

我个人推荐第二种方式。为什么？因为你可以针对不同的状态存储做差异化配置。比如聚合用的 store 写多，join 用的 store 读多，配置完全可以不一样。

3.3.4 监控与调优建议

调优不能靠猜，得有数据支撑。RocksDB 提供了丰富的监控指标：

• rocksdb.bytes.written：写入字节数，看写压力
• rocksdb.bytes.read：读取字节数，看读压力
• rocksdb.memtable.flush：MemTable 刷盘次数，刷太频繁说明 write_buffer_size 太小
• rocksdb.compaction.pending：等待 compaction 的任务数，积压说明 compaction 跟不上

我一般会把这些指标接入 Prometheus + Grafana。看到 compaction.pending 持续大于 0，我就知道该调大 max_background_compactions 了。