4

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

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);
    }
}

shouldReturnElapsedTime提取ContinuousCompletableFuture变量的第一个测试工作正常，但其他测试shouldOperateWithOwnExecutionTime失败。同时，我更喜欢在我未来的代码中看到它既不提取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方法使用。

有什么建议么？

4

1 回答 1

3

在你的方法中

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);
}
于 2015-10-12T13:46:04.140 回答