如何使可完成的未来在完成后通过垃圾回收进行回收？

Question

我基于java的CompletableFuture构建了一个任务链，它可能非常长。我的问题是CompletableFuture中的每一个任务都是一个内部类UniCompletion，它包含对源CompletableFuture的引用，因此不可能对已完成的CompletableForture进行垃圾收集。有没有办法避免内存泄漏？

这是一段可用于重现此错误的代码：

    public static void main(String... args) {
        ExecutorService executor = Executors.newSingleThreadExecutor();
        AtomicReference<CompletableFuture<Integer>> future = new AtomicReference<>(CompletableFuture.completedFuture(0));
        IntStream.range(0, 100000000).forEach(i -> future.set(future.get().thenApplyAsync(ii -> ii + 1, executor)));
        future.get().get();
        executor.shutdown();
        executor.awaitTermination(10, TimeUnit.SECONDS);
    }

Answer 1

当我使用以下程序时，

import java.lang.ref.Cleaner;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.locks.LockSupport;

public class CfGc {
    static final Cleaner CLEANER = Cleaner.create();
    static CompletableFuture<Integer> next(CompletableFuture<Integer> f) {
        Object[] status = { "not completed" };
        CLEANER.register(f, () -> System.out.println(status[0]+" future collected"));
        return f.whenComplete((i,t) -> {
                status[0] = t != null? t: i;
                LockSupport.parkNanos(500_000_000);
                System.out.println(status[0]+" completed, running gc()");
                System.gc();
                LockSupport.parkNanos(5_000_000);
                System.out.println(status[0]+" completed, gc() ran\n");
            }).thenApply(i -> i + 1);
    }
    public static void main(String[] args) {
        CompletableFuture<Integer> s = new CompletableFuture<>(), f = s;
        for(int i = 0; i < 6; i++) f = next(f);
        s.complete(1);
    }
}

它始终如一地在我的机器上打印

1 completed, running gc()
1 completed, gc() ran

2 completed, running gc()
2 completed, gc() ran

3 completed, running gc()
2 future collected
3 completed, gc() ran

4 completed, running gc()
3 future collected
4 completed, gc() ran

5 completed, running gc()
4 future collected
5 completed, gc() ran

6 completed, running gc()
5 future collected
6 completed, gc() ran

这表明，在评估未来的后续阶段时，未来是可以实现的，但在评估下一个阶段时，则无法实现。这不像整个链在最后一个阶段结束之前一直被引用。

只有第一个未来保持可访问性，这在链的顺序评估中是不可避免的，因为在第一个未来从主方法调用的完整方法中，一切都发生在完整方法中。当我们将程序更改为

static final Cleaner CLEANER = Cleaner.create();
static CompletableFuture<Integer> next(CompletableFuture<Integer> f) {
    Object[] status = { "not completed" };
    CLEANER.register(f, () -> System.out.println(status[0]+" future collected"));
    return f.whenComplete((i,t) -> {
            status[0] = t != null? t: i;
            LockSupport.parkNanos(500_000_000);
            System.out.println(status[0]+" completed, running gc()");
            System.gc();
            LockSupport.parkNanos(5_000_000);
            System.out.println(status[0]+" completed, gc() ran\n");
        }).thenApplyAsync(i -> i + 1);
}
public static void main(String[] args) {
    CompletableFuture<Integer> s = new CompletableFuture<>(), f = s;
    for(int i = 0; i < 6; i++) f = next(f);
    s.complete(1);
    s = null;
    f.join();
}

它打印

1 completed, running gc()
1 completed, gc() ran

2 completed, running gc()
1 future collected
2 completed, gc() ran

3 completed, running gc()
2 future collected
3 completed, gc() ran

4 completed, running gc()
3 future collected
4 completed, gc() ran

5 completed, running gc()
4 future collected
5 completed, gc() ran

6 completed, running gc()
5 future collected
6 completed, gc() ran

在我的机器上，展示了当在完成过程中没有从堆栈框架中引用初始future时，它也会被垃圾收集。

这同样适用于我们使用

public static void main(String[] args) {
    CompletableFuture<Integer> f = CompletableFuture.supplyAsync(() -> 1);
    for(int i = 0; i < 6; i++) f = next(f);
    f.join();
}

无论next方法是使用然后使用Apply，还是使用然后使用applyAsync。