java - 具有耗时信息的 CompletableFuture

Question

我需要访问有关异步方法执行时间的信息。所以，我正在尝试扩展 CompletableFuture功能。这是我用 decorator 实现的模式用法:

import java.util.concurrent.*;
import java.util.function.*;
import static lombok.AccessLevel.PRIVATE;
import lombok.AllArgsConstructor;
import lombok.experimental.Delegate;

@AllArgsConstructor(access = PRIVATE)
public class ContinuousCompletableFuture<T> extends CompletableFuture<T> {

    @Delegate
    private final CompletableFuture<T> baseFuture;

    private final long creationTime;

    public static <U> ContinuousCompletableFuture<U> supplyAsync(Supplier<U> supplier) {
        return new ContinuousCompletableFuture<>(CompletableFuture.supplyAsync(supplier));
    }

    private ContinuousCompletableFuture(CompletableFuture<T> baseFuture) {
        this.baseFuture = baseFuture;
        this.creationTime = System.nanoTime();
    }

    public Long getElapsedTime() {
        return (System.nanoTime() - creationTime) / 1000_000L;
    }

    public ContinuousCompletableFuture<Void> thenAcceptAsync(BiConsumer<? super T, Long> action) {
        CompletionStage<Long> elapsedTime = CompletableFuture.completedFuture(getElapsedTime());
        return new ContinuousCompletableFuture<>(baseFuture.thenAcceptBothAsync(elapsedTime, action), creationTime);
    }
}

首先使用提取的 ContinuousCompletableFuture 变量测试 shouldReturnElapsedTime 工作正常，但其他 shouldOperateWithOwnExecutionTime 失败。同时，我更愿意在我 future 的代码中看到它既不提取 ContinuousCompletableFuture 变量。

import java.util.concurrent.atomic.AtomicLong;
import lombok.extern.slf4j.Slf4j;
import org.junit.*;
import static org.junit.Assert.*;

@Slf4j
public class ContinuousCompletableFutureTest {

    private static final int DELAY = 1000;

    AtomicLong flag = new AtomicLong();

    ContinuousCompletableFuture<String> future;

    @Before
    public void before() {
        future = ContinuousCompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(DELAY);
            } catch (InterruptedException ex) {
                log.error("Error during ContinuousCompletableFuture execution", ex);
            }
            return "successfully completed";
        });
    }

    @Test
    public void shouldReturnElapsedTime() {
        future.thenAcceptAsync(s -> {
            long t = future.getElapsedTime();
            log.info("Elapsed {} ms to receive message \"{}\"", t, s);
            flag.set(t);
        });

        try {
            Thread.sleep(2000);
        } catch (InterruptedException ex) {
            log.error("Error awaiting Test completion", ex);
        }

        assertTrue("Future completion should be delayed", flag.get() >= 0.75 * DELAY);
    }

    @Test
    public void shouldOperateWithOwnExecutionTime() {
        future.thenAcceptAsync((s, t) -> {
            log.info("Elapsed {} ms to receive message \"{}\"", t, s);
            flag.set(t);
        });

        try {
            Thread.sleep(2000);
        } catch (InterruptedException ex) {
            log.error("Error awaiting Test completion", ex);
        }

        assertTrue("Future completion should be delayed", flag.get() >= 0.75 * DELAY);
    }
}

我认为我的问题在于错误的 thenAcceptBothAsync 方法使用。

有什么建议吗？

Answer 1

在你的方法中

public ContinuousCompletableFuture<Void> thenAcceptAsync(
                                         BiConsumer<? super T, Long> action) {
    CompletionStage<Long> elapsedTime=CompletableFuture.completedFuture(getElapsedTime());
    return new ContinuousCompletableFuture<>(
        baseFuture.thenAcceptBothAsync(elapsedTime, action), creationTime);
}

您正在立即评估 getElapsedTime() 并将结果原封不动地传递给 BiConsumer，而不管实际完成时间如何。

您可以通过直接在消费者中查询耗时来修复它:

public ContinuousCompletableFuture<Void> thenAcceptAsync(
                                         BiConsumer<? super T, Long> action) {
    return new ContinuousCompletableFuture<>(
        baseFuture.thenAcceptAsync(t -> action.accept(t, getElapsedTime())), creationTime);
}