如何在内存中的Kafka Streams状态存储上启用缓存

Question

我想减少下游发送的数据的数量，因为我只关心给定密钥的最后一个值，所以我正在以这种方式从主题中读取数据：

KTable table = build.table("inputTopic", Materialized.as("myStore"));

为什么？因为在引擎盖下，数据被缓存，如here所述，并且只有在commit.interval.ms或cache.max.bytes.buffering启动时才转发。

到目前为止这么好，但是因为在这种情况下我根本没有利用RocksDB，所以我想用内存存储的默认实现替换它。为了以防万一，我隐式启用了缓存。

Materialized.as(Stores.inMemoryKeyValueStore("myStore")).withCachingEnabled();

但是它不起作用 - 数据没有被缓存，每条记录都被发送到下游。

还有另一种启用缓存的方法吗？或者有更好的方法来做我想要实现的目标？

Answer 1

看来我错了，内存状态存储缓存按预期工作。我将简要介绍一下我是如何测试它的，也许有人会发现它很有用。我创建了一个非常基本的Kafka Streams应用程序，它只是从一个抽象为KTable的主题中读取。

public class Main {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        Logger logger = LoggerFactory.getLogger(Main.class);

        builder.table("inputTopic", Materialized.as(Stores.inMemoryKeyValueStore("myStore")).withCachingEnabled())
                .toStream()
                .foreach((k, v) -> logger.info("Result: {} - {}", k, v));

        new KafkaStreams(builder.build(), getProperties()).start();
    }

    private static Properties getProperties() {
        Properties properties = new Properties();
        properties.put(APPLICATION_ID_CONFIG, "testApp");
        properties.put(BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(COMMIT_INTERVAL_MS_CONFIG, 10000);
        properties.put(CACHE_MAX_BYTES_BUFFERING_CONFIG, 10 * 1024 * 1024L);
        properties.put(DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        properties.put(DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        return properties;
    }
}

然后我从Kafka运行控制台生产者：

/kafka-console-producer.sh --broker-list localhost:9092 --topic inputTopic --property "parse.key=true" --property "key.separator=:"

并发送了一些消息：a：a，a：b，a：c。在应用程序中只显示它们的最后一条消息，因此缓存按预期工作。

2018-03-06 21:21:57 INFO Main：26 - 结果：a - c

我也稍微改变了流来检查aggregate方法的缓存。

builder.stream("inputTopic")
        .groupByKey()
        .aggregate(() -> "", (k, v, a) -> a + v, Materialized.as(Stores.inMemoryKeyValueStore("aggregate")))
        .toStream()
        .foreach((k, v) -> logger.info("Result: {} - {}", k, v));

我使用相同的密钥快速连续发送了几条消息，而且我只收到了一个结果，所以数据没有立即发送到下游 - 正如预期的那样。