首页 > AI教程 >Flink闭包清除源码分析

Flink闭包清除源码分析

来源：互联网 2026-06-30 06:33:07

Flink算子通过序列化分发执行，匿名内部类引用外部对象形成闭包，若外部对象不可序列化则任务失败。框架使用ClosureCleaner工具，通过反射查找以this$开头的闭包引用字段，将其置为null，并借助ASM字节码验证是否需要清除，确保算子可正常序列化。

0x1 摘要

本文围绕 Flink 闭包清除机制，解答两个核心问题：为什么 Flink 需要执行闭包清除？以及 Flink 是如何具体实现闭包清除的？

Flink闭包清除源码分析

长期稳定更新的攒劲资源： >>>点此立即查看<<<

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

大家都知道，Flink 算子必须通过序列化才能分发到各个节点执行。因此算子的对象必须可序列化，这是基本要求。然而许多开发者为了方便，常使用匿名内部类实现算子——匿名内部类一旦引用外部对象，就会自动携带闭包。如果该外部对象没有实现 Serializable 接口，整个算子将因序列化失败而抛出异常。为了确保任务正常运行，Flink 框架必须在底层将这些闭包引用“清除”掉。

0x3 Flink 闭包清除实现

先看一个最简单的 Map 算子代码示例：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
final DataStreamSource source = env.addSource(new SourceFunction() {
    @Override
    public void run(SourceContext ctx) throws Exception {}
    @Override
    public void cancel() {}
});

source.map(new MapFunction() {
    @Override
    public String map(String value) throws Exception {
        return null;
    }
});

接下来跟进 map 方法的源码：

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

注意这里调用了 clean(mapper)。继续追查，最终会走到 StreamExecutionEnvironment 类的这个方法：

@Internal
public  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 字节码工具来检测闭包是否被实际访问，细节此处不展开，有兴趣可自行查阅源码。

侠游戏发布此文仅为了传递信息，不代表侠游戏网站认同其观点或证实其描述