Sprak中,线程之前的交互采用eventLoop模型。
当JobGenerate中的clock达到触发新一次job生成的时间后,并不会直接驱动graph去生成job,而是通过往eventLoop中发送一个JobGenerate事件以触发job生成事件的产生。
private val eventQueue: BlockingQueue[E] = new LinkedBlockingDeque[E]()
private val stopped = new AtomicBoolean(false)
// Exposed for testing.
private[spark] val eventThread = new Thread(name) {
setDaemon(true)
override def run(): Unit = {
try {
while (!stopped.get) {
val event = eventQueue.take()
try {
onReceive(event)
} catch {
case NonFatal(e) =>
try {
onError(e)
} catch {
case NonFatal(e) => logError("Unexpected error in " + name, e)
}
}
}
} catch {
case ie: InterruptedException => // exit even if eventQueue is not empty
case NonFatal(e) => logError("Unexpected error in " + name, e)
}
}
}
eventLoop维护了一个队列用来存放事件,由于队列的先进先出特性,导致可以按照时间顺序对相关事件进行处理,一个eventLoop也只维护了一个eventThread,将会不断循环往上文所述的队列尝试拉取事件,通过onReceive()方法,这里如果onReceive()方法的事件处理为同步处理,如果阻塞将会导致下一个事件的处理延时。
eventLoop除了在JobGenerate中用来触发job的生成外,还在JobScheduler中用来向消息总线汇报一个任务的启动与完成。
在JobScheduler中,所有job在准备启动前,都会被封装成一个JobHandler,在这个JobHandler的run()方法中,实现了job启动的生命周期行为,并在这个方法中每个job的行为都会通过eventLoop向消息总线报告其行为。
var _eventLoop = eventLoop
if (_eventLoop != null) {
_eventLoop.post(JobStarted(job, clock.getTimeMillis()))
// Disable checks for existing output directories in jobs launched by the streaming
// scheduler, since we may need to write output to an existing directory during checkpoint
// recovery; see SPARK-4835 for more details.
SparkHadoopWriterUtils.disableOutputSpecValidation.withValue(true) {
job.run()
}
_eventLoop = eventLoop
if (_eventLoop != null) {
_eventLoop.post(JobCompleted(job, clock.getTimeMillis()))
}
类比job,task的生命周期与消息总线的汇报也是通过eventLoop中的,由DAGScheduler实现。DAGScheduler中,job转换为stage这一最重要的步骤,也是通过eventLoop来投递JobSubmitted事件进行通知传递的。