我们是ApacheFlink和scala的新手。这里是我们的用例，就像我们从amps服务器（crankuptheamps）开始使用两种类型的主题来传播数据一样。源1和2从主题1和2中提取数据。
用例
我们的要求只是最初的数据源#1将在第二个数据源启动之前播送（世界现状）数据。因为我们只是将源1数据存储到Map状态。那么我们只需要启动source#2sow操作。最终我们需要逐个运行源代码。那么是否有任何选项可以逐个运行源代码。

def sourceConnect(environment: StreamExecutionEnvironment,topic: String, subscriptionType: SubscriptionType): DataStream[Map[String, String]] = {

val dataStream : DataStream[Map[String, String]] = environment.addSource(new RichSourceFunction[Map[String,String]]()  {

  var sourceClient: Client = null

  override def open(parameters: Configuration): Unit = {
    // .... Code Here
  }

  override def run(sourceContext: SourceFunction.SourceContext[Map[String, String]]): Unit = {
    subscriptionType match {

      case SubscriptionType.sow =>
          //.... Code Here

    }
  }

  override def getRuntimeContext: RuntimeContext = super.getRuntimeContext

  override def cancel(): Unit = {
    sourceClient.close()
  }

  override def close(): Unit = try cancel()
  finally super.close()

})
dataStream }

private var environment: StreamExecutionEnvironment = null

     // .... Code Here

     val source1 = environment.addSource(....)
     val source2 = environment.addSource(....)

     val conn = source1.connect(source2)

     conn.print()

     environment.execute()

最终我们的用例只需要先运行source1，然后运行source2，这意味着同步