Spark的广播变量是怎么实现的？——Java引用类型和堆外内存

// 四种引用的代码示例
 
// 强引用：最常见的赋值
Object obj = new Object();  // 只要obj变量存在，对象不会被GC
 
// 软引用：内存不够时才回收（适合缓存）
SoftReference<byte[]> cache = new SoftReference<>(new byte[10_000_000]);
byte[] data = cache.get();  // 如果还没被回收就能拿到
// 内存充足时data不为null，内存紧张时GC会回收它
 
// 弱引用：下次GC就回收
WeakReference<Object> weakRef = new WeakReference<>(new Object());
System.gc();
weakRef.get();  // 大概率是null了（已被GC回收）
 
// 虚引用：配合ReferenceQueue使用，对象回收时收到通知
ReferenceQueue<Object> queue = new ReferenceQueue<>();
PhantomReference<Object> phantom = new PhantomReference<>(obj, queue);
// obj被GC后，phantom会被放入queue
// 可以在queue中执行清理逻辑（比如释放堆外内存）

// 分配堆外内存
ByteBuffer directBuf = ByteBuffer.allocateDirect(1024 * 1024); // 1MB堆外
ByteBuffer heapBuf = ByteBuffer.allocate(1024 * 1024);         // 1MB堆内
 
// 堆外 vs 堆内
// 堆外：不受GC管理、可以零拷贝发送到网络、但分配/释放慢
// 堆内：受GC管理、需要拷贝到堆外才能发送网络、但分配/释放快
 
// 堆外内存的释放机制（虚引用+Cleaner）
// DirectByteBuffer内部有一个Cleaner对象（虚引用的子类）
// 当DirectByteBuffer对象被GC回收时
// Cleaner的clean()方法被调用，释放堆外内存
 
// 问题：如果堆上的DirectByteBuffer对象一直不被GC回收
//       堆外内存就一直不释放 ==> 堆外内存泄漏！
// 这就是为什么禁用System.gc()可能导致堆外内存问题
// 因为System.gc()是触发堆上对象回收从而释放堆外内存的手段

大数据框架的堆外内存使用

组件	堆外内存用途	配置参数	释放机制
Flink	网络缓冲、RocksDB缓存	taskmanager.memory.network	框架管理+Cleaner
Spark	Tungsten内存、Shuffle缓冲	spark.memory.offHeap.size	TaskMemoryManager管理
Kafka	零拷贝的FileChannel传输	socket.send/receive.buffer.bytes	OS管理
Doris BE	向量化计算的列缓冲	BE配置	BE内存管理器

实战踩坑：Spark广播变量的内存泄漏

在某知名出海硬件品牌，有一个Spark Streaming作业（微批处理），每个批次都创建一个新的广播变量来广播最新的维度表数据。但旧的广播变量没有调用unpersist()释放，导致Executor内存中堆积了大量旧版本的广播数据。

症状：Executor内存使用率持续上升，几个小时后开始频繁Full GC，最终OOM。

// 错误写法：每个批次创建新广播变量但不释放旧的
while (true) {
    Dataset<Row> dimTable = spark.read().jdbc(...);
    Broadcast<Dataset<Row>> bc = sc.broadcast(dimTable);
    // 处理当前批次...
    // 下一个批次又创建新的bc，旧的bc没有释放！
    // 旧bc的强引用还在Driver端，Executor上的数据不会被GC
}
 
// 正确写法：用完就释放
Broadcast<Dataset<Row>> prevBc = null;
while (true) {
    if (prevBc != null) prevBc.unpersist(true); // 释放旧广播
    Dataset<Row> dimTable = spark.read().jdbc(...);
    Broadcast<Dataset<Row>> bc = sc.broadcast(dimTable);
    // 处理当前批次...
    prevBc = bc;  // 记住当前广播，下轮释放
}

教训：Spark的广播变量底层使用强引用，不会被GC自动回收。每次创建新广播后必须手动unpersist旧广播。这在Streaming场景中特别容易被忽略——因为批处理作业只跑一次不需要关心释放，但Streaming是长期运行的。

面试高频追问

追问一：为什么不用弱引用来缓存广播变量？自动回收不好吗？

弱引用下次GC就回收——对于频繁访问的广播变量来说太快了，每次GC后都要重新传输广播数据，开销巨大。软引用更合适——只有内存真的不够时才回收。Spark的BlockManager内部确实用了软引用来缓存某些Block数据。但广播变量因为需要精确控制生命周期（新旧版本切换），用强引用+手动管理更可靠。

追问二：ThreadLocal为什么用弱引用？

ThreadLocal的内部Map用弱引用指向ThreadLocal对象本身（Key）。这样当ThreadLocal变量被赋null后（没有强引用了），下次GC就能回收Key，Map中的Entry变成key=null。但Value是强引用——必须手动调用remove()才能释放Value。不调remove()就会导致Value内存泄漏——这是ThreadLocal最常见的坑。Flink的ProcessFunction中如果用了ThreadLocal一定要在close()里remove。

追问三：怎么监控堆外内存使用？

堆外内存不在jmap/jstat的统计中，需要用其他手段：① NMT（Native Memory Tracking）：java -XX:NativeMemoryTracking=detail启动，用jcmd pid VM.native_memory查看；② Linux层面用pmap或/proc/pid/smaps看进程的虚拟内存映射；③ Flink/Spark的Web UI也会显示堆外内存使用量。

总结

引用类型面试讲清楚三层：① 四种引用的回收条件和典型用途（强=普通变量、软=缓存、弱=临时映射、虚=堆外资源释放）；② 堆外内存的分配和释放机制（DirectByteBuffer+Cleaner+虚引用）；③ 大数据场景的应用（Spark广播变量用强引用需手动释放、BlockManager用软引用做缓存、Flink网络缓冲用堆外内存）。能讲出Spark广播变量内存泄漏的踩坑和unpersist的修复方案，面试官会觉得你对内存管理有实战认知。

下一篇聊类加载机制：写Flink UDF遇到ClassNotFoundException——Java的双亲委派和Flink的ClassLoader是怎么回事？

你们遇到过堆外内存泄漏吗？怎么排查的？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长