Flink应用程序上的垃圾收集

Question

我在Scala中有一个非常简单的Flink应用程序。我有2个简单的流。我正在将我的一条小溪广播到另一个流。广播流包含规则，只是检查另一个是流的元组在规则内。一切正常，我的代码如下。

这是一个无限的运行应用程序。我想知道JVM是否有可能收集我的规则对象作为垃圾。

有人有任何想法吗？非常感谢。

object StreamBroadcasting extends App {
  val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
  val stream = env
    .socketTextStream("localhost", 9998)
    .flatMap(_.toLowerCase.split("\\W+").filter(_.nonEmpty))
    .keyBy(l => l)

  val ruleStream = env
    .socketTextStream("localhost", 9999)
    .flatMap(_.toLowerCase.split("\\W+").filter(_.nonEmpty))

  val broadcastStream: DataStream[String] = ruleStream.broadcast

  stream.connect(broadcastStream)
    .flatMap(new SimpleConnect)
    .print

  class SimpleConnect extends RichCoFlatMapFunction[String, String, (String, Boolean)] {
    private var rules: Set[String] = Set.empty[String] // Can JVM collect this object after a long time?

    override def open(parameters: Configuration): Unit = {}

    override def flatMap1(value: String, out: Collector[(String, Boolean)]): Unit = {
      out.collect(value, rules.contains(value))
    }

    override def flatMap2(value: String, out: Collector[(String, Boolean)]): Unit = {
      rules = rules.+(value)
    }
  }

  env.execute("flink-broadcast-streams")
}

Answer 1

不，一组规则不会收集垃圾。它将永远存在。 （当然，由于您不使用Flink的广播状态，因此规则将无法在申请中重新启动。）