引子:在使用spark和hadoop的时候,遇到一些进程退出时的报错。因此顺便研究了一下jvm以及一些开源框架的关闭钩子的机制。这篇文章不涉及底层native实现,仅限Java层面
1.jvm关闭钩子
注册jvm关闭钩子通过Runtime.addShutdownHook(),实际调用ApplicationShutdownHooks.add()。后者维护了一个钩子集合IdentityHashMap<Thread, Thread> hooks
ApplicationShutdownHooks类初始化的时候,会注册一个线程到Shutdown类
static { try { Shutdown.add(1, false, new Runnable() { public void run() { runHooks(); } } ); hooks = new IdentityHashMap<>(); } catch (IllegalStateException e) { hooks = null; } }
Shutdown类里也维护了一个钩子集合
private static final Runnable[] hooks = new Runnable[MAX_SYSTEM_HOOKS];
这个集合是分优先级的(优先级就是下标数值),自定义的钩子优先级默认是1,也就是最先执行。关闭钩子最终触发就是从这个集合进行
应用关闭时,以System.exit()为例,依次调用Runtime.exit()、Shutdow.exit()
Shutdown执行jvm退出逻辑,并维护了若干关闭状态
private static final int RUNNING = 0; // 初始状态,开始关闭 private static final int HOOKS = 1; // 运行钩子 private static final int FINALIZERS = 2; // 运行finalizer private static int state = RUNNING; static void exit(int status) { boolean runMoreFinalizers = false; synchronized (lock) { // 根据退出码status参数做不同处理 if (status != 0) runFinalizersOnExit = false; // 只有正常退出才会运行finalizer switch (state) { case RUNNING: // 执行钩子并修改状态 state = HOOKS; break; case HOOKS: // 执行钩子 break; case FINALIZERS: // 执行finalizer if (status != 0) { halt(status); // 如果是异常退出,直接退出进程。halt()底层是native实现,这时不会执行finalizer } else { // 正常退出则标记是否需要执行finalizer runMoreFinalizers = runFinalizersOnExit; } break; } } if (runMoreFinalizers) { // 如果有需要,就执行finalizer,注意只有state=FINALIZERS会走这个分支 runAllFinalizers(); halt(status); } synchronized (Shutdown.class) { // 这里执行state= HOOKS逻辑,包括执行钩子和finalizer sequence(); halt(status); } } private static void sequence() { synchronized (lock) { if (state != HOOKS) return; } runHooks(); // 执行钩子,这里会依次执行hooks数组里的各线程 boolean rfoe; // finalizer逻辑 synchronized (lock) { state = FINALIZERS; rfoe = runFinalizersOnExit; } if (rfoe) runAllFinalizers(); } private static void runHooks() { for (int i=0; i < MAX_SYSTEM_HOOKS; i++) { try { Runnable hook; synchronized (lock) { currentRunningHook = i; hook = hooks[i]; } // 由于之前注册了ApplicationShutdownHooks的钩子线程,这里又会回调ApplicationShutdownHooks.runHooks if (hook != null) hook.run(); } catch(Throwable t) { if (t instanceof ThreadDeath) { ThreadDeath td = (ThreadDeath)t; throw td; } } } } static void runHooks() { Collection<Thread> threads; synchronized(ApplicationShutdownHooks.class) { threads = hooks.keySet(); hooks = null; } // 注意ApplicationShutdownHooks里的钩子之间是没有优先级的,如果定义了多个钩子,那么这些钩子会并发执行 for (Thread hook : threads) { hook.start(); } for (Thread hook : threads) { try { hook.join(); } catch (InterruptedException x) { } } }
2.Spring关闭钩子
Spring在AbstractApplicationContext里维护了一个shutdownHook属性,用来关闭Spring上下文。但这个钩子不是默认生效的,需要手动调用ApplicationContext.registerShutdownHook()来开启
在自行维护ApplicationContext(而不是托管给tomcat之类的容器时),注意尽量使用ApplicationContext.registerShutdownHook()或者手动调用ApplicationContext.close()来关闭Spring上下文,否则应用退出时可能会残留资源
public void registerShutdownHook() { if (this.shutdownHook == null) { this.shutdownHook = new Thread() { @Override public void run() { // 这里会调用Spring的关闭逻辑,包括资源清理,bean的销毁等 doClose(); } }; // 这里会把spring的钩子注册到jvm关闭钩子 Runtime.getRuntime().addShutdownHook(this.shutdownHook); } }
3.Hadoop关闭钩子
Hadoop客户端初始化时,org.apache.hadoop.util.ShutdownHookManager会向Runtime注册一个钩子线程。ShutdownHookManager是一个单例类,并维护了一个钩子集合
private Set<HookEntry> hooks; static { Runtime.getRuntime().addShutdownHook( new Thread() { @Override public void run() { MGR.shutdownInProgress.set(true); // MGR是本类的单例 for (Runnable hook: MGR.getShutdownHooksInOrder()) { try { hook.run(); } catch (Throwable ex) { LOG.warn("ShutdownHook '" + hook.getClass().getSimpleName() + "' failed, " + ex.toString(), ex); } } } } ); }
这里HookEntry是hadoop封装的钩子类,HookEntry是带优先级的,一个priority属性。MGR.getShutdownHooksInOrder()方法会按priority依次(单线程)执行钩子
默认挂上的钩子就一个:org.apache.hadoop.fs.FileSystem$Cache$ClientFinalizer(priority=10),这个钩子用来清理hadoop FileSystem缓存以及销毁FileSystem实例。这个钩子是在第一次hadoop IO发生时(如FileSystem.get)lazy加载
此外调用FileContext.deleteOnExit()方法也会通过注册钩子
hadoop集群(非客户端)启动时,还会注册钩子清理临时路径
4.SparkContext关闭钩子
Spark也有关闭钩子管理类org.apache.spark.util.ShutdownHookManager,结构与hadoop的ShutdownHookManager基本类似
hadoop 2.x开始,spark的ShutdownHookManager会挂一个SparkShutdownHook钩子到hadoop的ShutdownHookManager(priority=40),用来实现SparkContext的清理逻辑。hadoop 1.x没有ShutdownHookManager,所以SparkShutdownHook直接挂在jvm上
def install(): Unit = { val hookTask = new Runnable() { // 执行钩子的回调进程,根据priority依次执行钩子 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).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(_) => // hadoop 1.x Runtime.getRuntime.addShutdownHook(new Thread(hookTask, "Spark Shutdown Hook")); } }
顺便说一下,hadoop的FileSystem实例底层默认是复用的,所以如果执行了两次fileSystem.close(),第二次会报错FileSystem Already Closed异常(即使表面上是对两个实例执行的)
一个典型的场景是同时使用Spark和Hadoop-Api,Spark会创建FileSystem实例,Hadoop-Api也会创建,由于底层复用,两者其实是同一个。因为关闭钩子的存在,应用退出时会执行两次FileSystem.close(),导致报错。解决这个问题的办法是在hdfs-site.xml增加以下配置,关闭FileSystem实例复用
<property> <name>fs.hdfs.impl.disable.cache</name> <value>true</value> </property>
5.总结
以下为相关调用逻辑整理
紫红色箭头表示钩子注册,蓝色箭头表示钩子触发
蓝色、黄色、红色线框分别表示Spring、Hadoop、Spark相关代码
相关推荐
该项目是针对大数据处理的一个毕业设计,主要利用了Scrapy爬虫框架和Hadoop生态系统来抓取、存储和分析海量的招聘信息。以下将详细介绍这两个技术框架及其在该项目中的应用。 **Scrapy框架** Scrapy是一个开源的...
与Hadoop和Spark不同,MPI并不提供内置的分布式存储,用户需要自定义数据管理策略。 【TCP连接建立过程】 TCP连接建立通常经历三次握手(TCP Three-Way Handshake)。首先,客户端发送一个SYN(Synchronize ...
5. **API 和集成**:Atlas 提供 RESTful API,使得其他系统能够轻松地与之交互,如 Hive、Hadoop、Spark 等。Hive Hook 就是这样的一个集成示例,它使得 Hive 与 Atlas 之间的通信成为可能。 6. **安全性**:Apache...
我们可以使用Apache Hadoop、Spark等大数据处理框架来处理大规模数据。Hadoop提供分布式存储(HDFS)和处理(MapReduce)能力,而Spark则以其内存计算和流处理能力提高了数据处理速度。通过API接口或ETL(提取、转换...
Hadoop、Spark、MPI计算框架特点及适用场景: 1. Hadoop是一个开源框架,用于分布式存储和处理大数据。它包含两个主要部分:HDFS(用于存储)、MapReduce(用于处理)。Hadoop适用于需要处理大量数据,且对数据处理...
5. **集成**:Apache Atlas 集成了许多大数据生态系统中的其他组件,如 Hive、Hadoop、Spark 等,以便在整个数据生命周期中实现元数据的统一管理。 在这个 "apache-atlas-2.2.0-falcon-hook" 中,"Falcon" 是另一个...
CDH 6.3.1 是 Cloudera 的一个主要版本,它是一个企业级的大数据平台,包含 Hadoop、Spark、Impala 等组件。在这个版本中,Impala Hook 与 Atlas 的集成意味着用户可以在进行实时分析的同时,享受到 Atlas 提供的...
综上所述,"apache-atlas-2.1.0-hive-hook.tar.gz" 包含了 Apache Atlas 的 Hive 钩子组件,用于在 CDH 6.3.1 环境中实现高效的数据管理和治理。这个压缩包的解压后文件 "apache-atlas-hive-hook-2.1.0" 将包含所有...
`apache-atlas-2.2.0-hbase-hook.tar.gz` 文件是Apache Atlas 2.2.0版本针对HBase数据库的钩子(hook)组件,它允许Atlas与HBase紧密集成,提供更高效的数据管理和监控。 Apache HBase是一个非关系型分布式列式...