如何扩展 ReferenceQueue/Cleaner 的清理处理

2026-11-04技术老齐

结论

仅凭 ReferenceQueue 无法判断消费者是否已经落后，也无法保证积压量相对于已创建对象数量为 o(n)。Java 不会公开“已经 phantom reachable，但应用尚未处理”的准确数量。GC 策略还会影响引用的发现和入队时间，应用无法依赖其及时性。

可以将处理分成两层：

ReferenceQueue 消费者只做快速、固定成本的登记注销。
将耗时的 Huge.reduceIfNeeded() 交给有界且能够合并任务的调度器。
每个 Huge 维护增量统计、滞回阈值、最小回收批量和冷却时间，减少扫描和重复缩容。
监控应用能够准确观测的工作队列，不要推测 JVM 内部的引用积压。
如果内存必须在确定时间内释放，应提供 close() 之类的显式生命周期。PhantomReference 或 Cleaner 只负责兜底。

单纯增加 ReferenceQueue 消费线程通常解决不了问题。只要出队处理保持在 O(1) 或近似 O(1)，一个消费者一般就够了。真正需要控制容量和并行度的是后续的耗时处理。

为什么无法从计数推导队列长度

设：

C：创建的对象数量
D：已经从 ReferenceQueue 取出并处理的数量
L：仍然存活的对象数量
P：已被 GC 发现并等待处理的对象数量

现有计数只能说明：

C - D = L + P

L 是未知的，因此无法通过 C - D 算出 P。GC 何时判定对象为 phantom reachable，以及何时将引用入队，也没有应用层可以依赖的时间上界。

注册时间与出队时间之差也不等于队列等待时间，因为其中包含了三个阶段：

对象正常存活时间
+ 等待 GC 发现的时间
+ ReferenceQueue 中等待消费的时间

连续 poll() 成功的次数和 remove() 的阻塞时长，只能反映消费者最近观察到的活动，不能准确表示尚未处理的引用总数。线程 CPU time 也只能说明线程消耗了多少 CPU，无法补充 JVM 未公开的状态。

假设引用产生的长期平均速率为 λ，消费者的处理能力为 μ。只有长期满足 λ < μ，并且批量峰值有限，系统才有可能恢复到没有积压的状态。ReferenceQueue API 无法保证这些条件，也无法给出 o(n) 的空间界限。

简化 ReferenceQueue 消费者

出队线程不应执行结构扫描、数组复制、压缩等耗时操作。它只处理以下工作：

从 Huge 的跟踪集合中删除对应的 PhantomReference
更新计数或增量摘要
将该 Huge 标记为 dirty
合并并调度后续检查
解除 PhantomReference 持有的无用状态

消费循环可以先阻塞等待一个引用，再用 poll() 批量取出当前已经入队的引用：

final class ReferenceConsumer implements Runnable {
    private static final int MAX_DRAIN_BATCH = 1024;

    private final ReferenceQueue<Lightweight> queue;

    ReferenceConsumer(ReferenceQueue<Lightweight> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                LightweightState first =
                        (LightweightState) queue.remove();

                process(first);

                for (int i = 1; i < MAX_DRAIN_BATCH; i++) {
                    LightweightState next =
                            (LightweightState) queue.poll();

                    if (next == null) {
                        break;
                    }
                    process(next);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    private void process(LightweightState state) {
        Huge owner = state.owner();
        owner.onCollected(state);
        state.clear();
    }
}

限制单批数量，可以防止持续到来的引用占满处理时间，使其他维护工作长期得不到执行。具体上限是运行参数，没有适合所有负载的固定值。

如果 onCollected() 仍需争用全局锁、修改复杂索引或执行线性扫描，就应继续拆分其中的工作，增加出队线程并不能消除这些成本。

在 Huge 内合并缩容请求

每个 Huge 可以维护这些状态：

当前注册的引用数
自上次检查以来注销的引用数
可以增量维护的使用范围摘要
是否已经安排缩容检查
上次实际缩容的时间和容量
当前结构容量及估算的必要容量

下面是一个调度骨架。阈值和容量计算仍需按照实际数据结构定义：

final class Huge {
    private final Set<LightweightState> lightweights =
            ConcurrentHashMap.newKeySet();

    private final AtomicLong removedSinceCheck = new AtomicLong();
    private final AtomicBoolean checkScheduled = new AtomicBoolean();

    private final ScheduledExecutorService scheduler;
    private final Executor resizeExecutor;

    Huge(
            ScheduledExecutorService scheduler,
            Executor resizeExecutor) {
        this.scheduler = scheduler;
        this.resizeExecutor = resizeExecutor;
    }

    void register(LightweightState state) {
        lightweights.add(state);
        addToUsageSummary(state.projection());
    }

    void onCollected(LightweightState state) {
        if (!lightweights.remove(state)) {
            return;
        }

        removeFromUsageSummary(state.projection());
        removedSinceCheck.incrementAndGet();

        if (cheapEligibilityTest()) {
            scheduleCheck();
        }
    }

    private boolean cheapEligibilityTest() {
        return removedSinceCheck.get() >= minimumRemovalBatch()
                && cooldownExpired();
    }

    private void scheduleCheck() {
        if (!checkScheduled.compareAndSet(false, true)) {
            return;
        }

        scheduler.schedule(
                () -> resizeExecutor.execute(this::runReductionCheck),
                debounceDelay(),
                debounceTimeUnit());
    }

    private void runReductionCheck() {
        long observed = removedSinceCheck.get();

        try {
            if (shouldReduce()) {
                reduceIfNeeded();
            }
            removedSinceCheck.addAndGet(-observed);
        } finally {
            checkScheduled.set(false);

            /*
             * 处理检查期间又有对象被注销的情况。
             * scheduleCheck() 内的 CAS 会合并并发请求。
             */
            if (removedSinceCheck.get() > 0 && cheapEligibilityTest()) {
                scheduleCheck();
            }
        }
    }

    private boolean shouldReduce() {
        long required = estimateRequiredCapacity();
        long capacity = currentCapacity();

        return required < shrinkTrigger(capacity);
    }

    private void reduceIfNeeded() {
        long required = computeRequiredCapacityPrecisely();
        long target = targetCapacityWithHeadroom(required);

        if (target < currentCapacity()) {
            resizeTo(target);
            recordResize();
        }
    }

    // 以下方法由具体数据结构实现。
    private void addToUsageSummary(Projection p) {}
    private void removeFromUsageSummary(Projection p) {}
    private long minimumRemovalBatch() { throw new UnsupportedOperationException(); }
    private boolean cooldownExpired() { throw new UnsupportedOperationException(); }
    private long debounceDelay() { throw new UnsupportedOperationException(); }
    private TimeUnit debounceTimeUnit() { throw new UnsupportedOperationException(); }
    private long estimateRequiredCapacity() { throw new UnsupportedOperationException(); }
    private long computeRequiredCapacityPrecisely() { throw new UnsupportedOperationException(); }
    private long currentCapacity() { throw new UnsupportedOperationException(); }
    private long shrinkTrigger(long capacity) { throw new UnsupportedOperationException(); }
    private long targetCapacityWithHeadroom(long required) { throw new UnsupportedOperationException(); }
    private void resizeTo(long target) {}
    private void recordResize() {}
}

示例中的计数更新还要根据并发模型补充。例如，能否使用 addAndGet(-observed)，取决于其他路径是否可能重置该计数。需要保留的状态转换如下：

多个引用注销
    -> 一个 Huge 变为 dirty
    -> 最多安排一个延迟检查
    -> 检查期间继续积累变更
    -> 如有必要再安排下一次检查

使用滞回来控制缩容

如果缩容和停止缩容共用一个容量阈值，当使用量在边界附近波动时，大型结构很容易被反复复制。

可以分别设置触发参数和目标参数：

触发条件：
requiredCapacity < currentCapacity × triggerRatio

缩容目标：
targetCapacity >= requiredCapacity × headroomFactor

同时至少要加入一种摊销条件：

自上次检查以来已有足够多的对象被注销
预计可回收的绝对空间达到下限
预计可回收空间占当前容量的比例达到下限
距离上次缩容已经超过冷却时间
延迟一小段时间，合并同一批 GC 产生的注销事件

这些参数没有统一的正确值，需要根据复制成本、内存压力和典型对象批次确定。

如果 Projection 表示重叠区间，直接累加各个视图的使用量通常会重复计算。可以维护区间引用计数、分段计数、最大端点的有序多重集合，或者符合实际访问模型的摘要结构。注销单个视图时只更新摘要，完整计算则留到缩容条件初步成立后再执行。

应该监控哪些指标

应当监控应用能够控制和观测的处理阶段：

ReferenceQueue 每秒实际取出的引用数
每批取出的引用数
onCollected() 的耗时分布
dirty Huge 的数量
已调度但尚未执行的缩容检查数量
最老缩容任务的等待时间
缩容执行时间和实际释放的容量
被合并的重复调度次数
因冷却时间或回收量不足而跳过缩容的次数
有界执行器的拒绝次数或饱和时间

其中，“最老待执行任务的年龄”通常比瞬时队列长度更能说明后端是否长期处理不过来。

执行器如果需要排队，应使用有界队列。无界执行器只会把 ReferenceQueue 的积压转移成另一个更隐蔽的内存问题。执行器饱和时，可以选择以下处理方式：

保留 dirty 标志，由周期性任务重新提交
由提交线程执行轻量检查，同时避免在引用消费者中执行耗时缩容
按 Huge 去重，每个实例最多保留一个待执行任务
将多个 dirty 实例放入有界的去重工作集合

任务被拒绝后不能直接丢弃，否则对应的 Huge 可能再也得不到检查。

什么时候增加 ReferenceQueue 消费者

只有测量结果表明出队阶段本身已经成为瓶颈，才有必要增加消费者。可以依据以下现象判断：

队列消费者长期几乎不阻塞
出队速率已经达到稳定上限
onCollected() 已经缩减为短小的固定成本操作
CPU 分析显示时间确实消耗在出队或注销阶段
增加消费者后吞吐量确有提升，也没有引发更严重的锁竞争

多个线程可以同时对一个 ReferenceQueue 调用 remove()，但会产生额外竞争。同一个 Huge 的引用也可能由不同线程同时处理，因此所有实例级状态都要支持并发更新。

一般可以采用这样的组合：

一个或少量固定的 ReferenceQueue 消费者
使用 poll() 批量取出引用
按 Huge 合并后续任务
使用独立且有界的缩容执行器

动态调整消费者数量或线程优先级，不适合作为主要控制手段。线程优先级只是调度提示，无法保证容量和及时性。

使用 Cleaner 时采用相同原则

Cleaner 可以简化 PhantomReference 和 ReferenceQueue 的管理，但它不会提供待清理任务数量，也不保证任务在指定时间内完成。清理动作应尽快返回。

清理动作可以只负责轻量注销和调度：

final class CleanupAction implements Runnable {
    private final Huge owner;
    private final Projection projection;

    CleanupAction(Huge owner, Projection projection) {
        this.owner = owner;
        this.projection = projection;
    }

    @Override
    public void run() {
        owner.onViewCollected(projection);
    }
}

注册时必须确保 CleanupAction 没有直接或间接引用已注册的 Lightweight：

final class Lightweight implements AutoCloseable {
    private static final Cleaner CLEANER = Cleaner.create();

    private final State state;
    private final Cleaner.Cleanable cleanable;

    Lightweight(Huge huge, Projection projection) {
        this.state = new State(huge, projection);
        this.cleanable = CLEANER.register(
                this,
                new CleanupAction(huge, projection));
    }

    @Override
    public void close() {
        cleanable.clean();
    }
}

如果清理动作通过 State、lambda、非静态内部类等路径重新引用 Lightweight，对象可能永远无法进入预期的清理流程。清理动作最好使用静态类，并逐项检查对象之间的引用关系。

创建多个 Cleaner 并按 Huge 分片可以隔离任务，但会增加线程和管理成本，也无法合并耗时的缩容操作。即使改用 Cleaner，缩容仍应交给独立的去重调度器。

显式生命周期更可靠

如果结构未能及时缩减可能导致应用耗尽内存，那么只依靠引用处理并不安全。GC 可能延后运行，引用入队和清理线程也没有完成期限。

可以提供这样的接口：

try (Lightweight view = huge.openView(projection)) {
    view.method();
}

close() 负责确定性地注销使用关系，Cleaner 或 PhantomReference 用于处理调用方遗漏 close() 的情况。清理逻辑必须幂等，以免显式关闭和 GC 兜底重复执行。

如果对象方法读取 state 后继续在 state 或 Huge 上操作，还需注意过早可达性分析。必要时，可以在操作结束处调用：

Reference.reachabilityFence(this);

这样可以防止相关操作尚未结束时，Lightweight 就被提前视为不可达。

实际落地建议

先把引用消费者的职责缩减为删除、计数、标记和调度。确认这条路径不包含线性扫描后，用一个消费者进行压测，然后重点观察缩容执行器的任务年龄、饱和情况以及每次缩容实际释放的容量。

如果后端出现积压，可以先合并更多缩容任务、改进增量摘要，或者调整缩容并行度。只有分析确认出队阶段确实达到处理上限时，再增加固定数量的 ReferenceQueue 消费者。

这套方案不能证明 JVM 内部队列为 o(n)。它能够避免应用层为每个被回收的视图创建一项耗时任务，并为应用可控的处理阶段设定容量、背压和监控边界。