您现在的位置是:首页 > 文章详情

Flink 闭包清除源码分析

日期:2019-06-05点击:684

0x1 摘要

本文主要讲解Flink里为什么需要做闭包清除?Flink是怎么实现闭包清除的?

0x2 Flink 为什么要做闭包清除

大家都知道Flink中算子都是通过序列化分发到各节点上,所以要确保算子对象是可以被序列化的,很多时候大家比较喜欢直接用匿名内部类实现算子,而匿名内部类就会带来闭包问题,当匿名内部类引用的外部对象没有实现序列化接口时,就会导致内部类无法被序列化,因此Flink框架底层必须做好清除工作。

0x3 Flink 闭包清除实现

先来看一个Map算子代码:

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); final DataStreamSource<String> source = env.addSource(new SourceFunction<String>() { @Override public void run(SourceContext<String> ctx) throws Exception { } @Override public void cancel() { } }); source.map(new MapFunction<String, String>() { @Override public String map(String value) throws Exception { return null; } });

跟进源码查看map方法:

public <R> SingleOutputStreamOperator<R> map(MapFunction<T, R> mapper) { TypeInformation<R> outType = TypeExtractor.getMapReturnTypes(clean(mapper), getType(), Utils.getCallLocationName(), true); return transform("Map", outType, new StreamMap<>(clean(mapper))); }

重点关注clean(mapper)代码,继续跟进源码,最终会走到StreamExecutionEnvironment类的以下方法:

@Internal public <F> F clean(F f) { if (getConfig().isClosureCleanerEnabled()) { ClosureCleaner.clean(f, true); } ClosureCleaner.ensureSerializable(f); return f; }

到这里已经可以看出来闭包清除工具类ClosureCleaner,下面我们详细剖析一下此类。
先看clean方法:

public static void clean(Object func, boolean checkSerializable) { if (func == null) { return; } final Class<?> cls = func.getClass(); // First find the field name of the "this$0" field, this can // be "this$x" depending on the nesting boolean closureAccessed = false; for (Field f: cls.getDeclaredFields()) { if (f.getName().startsWith("this$")) { // found a closure referencing field - now try to clean closureAccessed |= cleanThis0(func, cls, f.getName()); } } if (checkSerializable) { try { InstantiationUtil.serializeObject(func); } catch (Exception e) { String functionType = getSuperClassOrInterfaceName(func.getClass()); String msg = functionType == null ? (func + " is not serializable.") : ("The implementation of the " + functionType + " is not serializable."); if (closureAccessed) { msg += " The implementation accesses fields of its enclosing class, which is " + "a common reason for non-serializability. " + "A common solution is to make the function a proper (non-inner) class, or " + "a static inner class."; } else { msg += " The object probably contains or references non serializable fields."; } throw new InvalidProgramException(msg, e); } } }

方法参数:

  • func:要清除的对应
  • checkSerializable:清除完成后是否需要调用序列方法进行验证

第一步:查找闭包引用的成员变量,通过反射判断成员变量名是否包含this$来判定,代码片断:

for (Field f: cls.getDeclaredFields()) { if (f.getName().startsWith("this$")) { // found a closure referencing field - now try to clean closureAccessed |= cleanThis0(func, cls, f.getName()); } }

找到闭包引用的成员变量后,调用内部私有方法cleanThis0方法处理,看方法源码:

private static boolean cleanThis0(Object func, Class<?> cls, String this0Name) { This0AccessFinder this0Finder = new This0AccessFinder(this0Name); getClassReader(cls).accept(this0Finder, 0); final boolean accessesClosure = this0Finder.isThis0Accessed(); if (LOG.isDebugEnabled()) { LOG.debug(this0Name + " is accessed: " + accessesClosure); } if (!accessesClosure) { Field this0; try { this0 = func.getClass().getDeclaredField(this0Name); } catch (NoSuchFieldException e) { // has no this$0, just return throw new RuntimeException("Could not set " + this0Name + ": " + e); } try { this0.setAccessible(true); this0.set(func, null); } catch (Exception e) { // should not happen, since we use setAccessible throw new RuntimeException("Could not set " + this0Name + " to null. " + e.getMessage(), e); } } return accessesClosure; }

核心代码this0.set(func, null);将闭包引用置空处理,此方法还用到了ASM包,具体逻辑没完成整明白。

原文链接:https://yq.aliyun.com/articles/704755
关注公众号

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。

持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。

文章评论

共有0条评论来说两句吧...

文章二维码

扫描即可查看该文章

点击排行

推荐阅读

最新文章