如何将 KStream 的 Object 类型的值存储在 Hashmap 中？答案

【问题标题】：How to store the Value of Object type of a KStream in a Hashmap?如何将 KStream 的 Object 类型的值存储在 Hashmap 中？
【发布时间】：2020-05-18 17:29:04
【问题描述】：

我有一个像KStream<String, Object> 这样的 KStream 对象。我想在 HashMap 中获取 Object 的值，并在不同的 KStream 中处理后将其存储为值？

例如，

KStream key = "First"

KStream 值 = {id=1, name="abc" age=12}

期望的结果： HashMap hm ={id=1, name="abc" age=12}

【问题讨论】：

您的 json sn-p 中似乎缺少某些内容。否则你的意思是 KStream 值单独产生 HashMap 的键和值？
是的，Json 包含一个 java 对象，所以我有 hashmap 的键和值。

标签： java apache-kafka hashmap apache-kafka-streams spring-kafka

【解决方案1】：

您似乎对概念有一些误解。

KStream 是对主题（以及其中的所有记录）的逻辑抽象。它允许您一一处理这些记录。例如，您可以这样做：

KStream stream = builder.stream(...);
stream.foreach((k, v) -> { /* your processing logic */});

我想在 HashMap 中获取 Object 的值

您似乎希望在与KStream 相同的范围内拥有HashMap。但是，您不能从“外部”访问任何记录，只能通过将运算符（如上例所示的foreach()）应用于KStream 以将KStream 转换为新的`KStream。

在不同的 KStream 中处理后将其存储为值

注意确定这是什么意思。您可以通过在其上应用运算符将KStream 转换为新的KStream。例如：

KStream result = kstream.map(...);

此程序将为kstream 中的每条记录执行您传递给map() 的函数，并将映射函数的结果记录附加到result。即，map() 将每个输入记录转换为新的输出记录。

注意，当你定义你的处理时，你基本上指定了一个数据流程序，在你调用KafkaStreams#start()之后会持续执行。

【讨论】：