简单的KTable构造

Question

概念上的

假设我有一个名为“address”的主题，其中键是人名(String)，值是person的地址(String)。对个人地址的更新将是由他们的姓名作为密钥和新地址作为值组成的消息。所以，如果我只想要任何一个键的最近值，我想我会做一个ktable。当我这么做的时候，这里到底发生了什么？卡夫卡是否创造了一个新的话题，实际上是ktable和截断旧的价值？还是我必须为当前的地址创建一个新的主题？还是完全是别的什么？

实战

我发现的所有示例和教程都使用了不推荐的方法，所以我希望能找到一些更新的方法。我目前的解决方案是这样阅读这个主题：

final Properties config = new Properties();
//leaving out all the config.put() for readability

final Consumer<String, String> consumer = new KafkaConsumer<>(config);

consumer.subscribe(Collections.singletonList("addresses"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(200);
        for (ConsumerRecord<String, String> record : records) {
            //do stuff
        }
    }
} finally {
    consumer.close();
}

这起作用了一段时间，但现在我想要一个有状态的解决方案。有什么简单的方法可以在信息进来的时候将信息放入ktable中呢？我不想过滤它或任何东西，我只是想要任何条目来更新状态。提前谢谢。

Answer 1

KTable不创建新主题。相反，它将源主题上的消息视为数据库中的重新插入--如果这是第一次看到密钥，就像插入一个新记录。如果这不是第一次，它将被视为对现有记录的更新。KTable可以用作卡夫卡流的其他部分的输入，您也可以将该状态保存在本地存储中，并在应用程序的其他部分查询K/V存储。