scala - Kafka 流与 Scala

Question

我正在尝试将 kafka 流与 scala 一起使用下面是我在 Java 中的代码，它工作得很好:

KStreamBuilder builder = new KStreamBuilder();
    KStream<String, String> textLines = builder.stream("TextLinesTopic");
    textLines.foreach((key,values) -> {
        System.out.println(values);
    });

    KafkaStreams streams = new KafkaStreams(builder, config);
    streams.start();

我的scala代码如下:

  val builder = new KStreamBuilder
  val textLines:KStream[String, String]  = builder.stream("TextLinesTopic")
  textLines.foreach((key,value)-> {
   println(key)
  })

  val streams = new KafkaStreams(builder, config)
  streams.start()

Scala 代码抛出编译错误。预期类型不匹配:ForEachAction[>String,>String],Actual((any,any), Unit)未找到:值键未找到:值值

有谁知道如何在 scala 中使用 streams API

Answer 1

你的语法错误 :)。 -> 只是创建对的运算符，所以表达式

(key,value)-> {
  println(key)
}

具有类型 ((Any, Any), Unit) 因为编译器无法推断任何类型信息(并且缺少 key 和 value)

如果您使用的是 scala 2.12，将 -> 替换为 => 应该可以解决问题，但是如果您使用的是旧版本的 scala，则必须实现 java明确的双功能:

 textLines.foreach(new BiFunction[T1, T2] { ... })