Spark Streaming程序的停止可以是强制停止、异常停止或其他方式停止。
首先我们看StreamingContext的stop()方法
def stop( stopSparkContext: Boolean = conf.getBoolean("spark.streaming.stopSparkContextByDefault", true)
): Unit = synchronized {
stop(stopSparkContext, false)
}这里定义了两个参数，stopSparkContext可以通过配置文件定义，接着看接收两个参数的stop方法，代码如下
/**
* Stop the execution of the streams, with option of ensuring all received data
* has been processed.
*
* @param stopSparkContext if true, stops the associated SparkContext. The underlying SparkContext
* will be stopped regardless of whether this StreamingContext has been
* started.
* @param stopGracefully if true, stops gracefully by waiting for the processing of all
* received data to be completed
*/def stop(stopSparkContext: Boolean, stopGracefully: Boolean): Unit = { var shutdownHookRefToRemove: AnyRef = null
if (AsynchronousListenerBus.withinListenerThread.value) { throw new SparkException("Cannot stop StreamingContext within listener thread of" + " AsynchronousListenerBus")
}
synchronized { try {
state match { case INITIALIZED =>
logWarning("StreamingContext has not been started yet") case STOPPED =>
logWarning("StreamingContext has already been stopped") case ACTIVE =>
scheduler.stop(stopGracefully) // Removing the streamingSource to de-register the metrics on stop()
env.metricsSystem.removeSource(streamingSource)
uiTab.foreach(_.detach())
StreamingContext.setActiveContext(null)
waiter.notifyStop() if (shutdownHookRef != null) {
shutdownHookRefToRemove = shutdownHookRef
shutdownHookRef = null
}
logInfo("StreamingContext stopped successfully")
}
} finally { // The state should always be Stopped after calling `stop()`, even if we haven't started yet
state = STOPPED
}
} if (shutdownHookRefToRemove != null) {
ShutdownHookManager.removeShutdownHook(shutdownHookRefToRemove)
} // Even if we have already stopped, we still need to attempt to stop the SparkContext because
// a user might stop(stopSparkContext = false) and then call stop(stopSparkContext = true).
if (stopSparkContext) sc.stop()
}注释中说明要停止程序时，正确的方式是需要所有接收的数据被处理完成后再停止，那么就需要我们传入的stopGracefully参数为true，然后停止时会等待所有任务执行完成
Spark Streaming提供了一个优雅停止的方法，在StreamingContext里面有一个stopOnShutdown()方法，代码如下
private def stopOnShutdown(): Unit = {
val stopGracefully = conf.getBoolean("spark.streaming.stopGracefullyOnShutdown", false)
logInfo(s"Invoking stop(stopGracefully=$stopGracefully) from shutdown hook") // Do not stop SparkContext, let its own shutdown hook stop it
stop(stopSparkContext = false, stopGracefully = stopGracefully)
}stopOnShutdown()方法是什么意思呢，在我们的程序退出时，不管是正常退出或异常退出，stopOnShutdown()方法都会被回调，然后调用stop方法。stopGracefully 可以通过配置项spark.streaming.stopGracefullyOnShutdown配置，生产环境需要配置为true.
stopOnShutdown()方法是怎样被调用的呢？在StreamingContext的start方法中有一行代码
shutdownHookRef = ShutdownHookManager.addShutdownHook(StreamingContext.SHUTDOWN_HOOK_PRIORITY)(stopOnShutdown)
添加stopOnShutdown函数到ShutdownHookManager中，addShutdownHook代码如下
def addShutdownHook(priority: Int)(hook: () => Unit): AnyRef = {
shutdownHooks.add(priority, hook)
}看SparkShutdownHookManager 里都有什么，看代码注释了解SparkShutdownHookManager的功能，不一一介绍
private [util] class SparkShutdownHookManager { // 优先级队列，优先级越大，越优先执行
private val hooks = new PriorityQueue[SparkShutdownHook]() @volatile private var shuttingDown = false
/**
* Install a hook to run at shutdown and run all registered hooks in order. Hadoop 1.x does not
* have `ShutdownHookManager`, so in that case we just use the JVM's `Runtime` object and hope for
* the best.
*/
// 这里实例化一个线程，添加到jvm的关闭钩子中，等到jvm退出时才会被调用
def install(): Unit = {
val hookTask = new Runnable() { override def run(): Unit = runAll()
} Try(Utils.classForName("org.apache.hadoop.util.ShutdownHookManager")) match { case Success(shmClass) =>
val fsPriority = classOf[FileSystem]
.getField("SHUTDOWN_HOOK_PRIORITY")
.get(null) // static field, the value is not used
.asInstanceOf[Int]
val shm = shmClass.getMethod("get").invoke(null)
shm.getClass().getMethod("addShutdownHook", classOf[Runnable], classOf[Int])
.invoke(shm, hookTask, Integer.valueOf(fsPriority + 30)) case Failure(_) =>
Runtime.getRuntime.addShutdownHook(new Thread(hookTask, "Spark Shutdown Hook"));
}
} // jvm退出时钩子回调此函数
def runAll(): Unit = {
shuttingDown = true
var nextHook: SparkShutdownHook = null
//循环从优先级队列取数据执行，优先级越大，越优先执行
while ({ nextHook = hooks.synchronized { hooks.poll() }; nextHook != null }) {
Try(Utils.logUncaughtExceptions(nextHook.run()))
}
} def add(priority: Int, hook: () => Unit): AnyRef = {
hooks.synchronized { if (shuttingDown) { throw new IllegalStateException("Shutdown hooks cannot be modified during shutdown.")
}
val hookRef = new SparkShutdownHook(priority, hook)
hooks.add(hookRef)
hookRef
}
} def remove(ref: AnyRef): Boolean = {
hooks.synchronized { hooks.remove(ref) }
}
}看到这里就明白了，把stopOnShutdown()函数放入SparkShutdownHookManager 中的优化级队列hooks中，默认优先级为51，jvm退出时启动一个线程，调用runAll()方法，然后从hooks队列中一个一个取数据(函数)，然后执行，就调用了stopOnShutdown()函数，接着调用stop()函数，我们的应用程序就可以优雅的执行停止工作了。
作者：海纳百川_spark
链接：https://www.jianshu.com/p/18cd94b5c647