Java 8 CompletedFuture 和 ThreadLocalRandom

Question

我正在练习 Cay S. Horstmann 所著的“真正不耐烦的 Java SE 8”一书。

写一个方法

public static <T> CompletableFuture<T> repeat(
         Supplier<T> action, Predicate<T> until)

异步重复操作，直到它产生一个值 被until 函数接受，它也应该运行 异步。使用读取 a 的函数进行测试 java.net.PasswordAuthentication 来自控制台，还有一个函数 通过睡眠一秒钟然后模拟有效性检查 检查密码是否“秘密”。

我想出了以下代码，但随机密码生成策略似乎让我失望了。所有线程不断选择相同的密码，这看起来很奇怪。

public static <T> CompletableFuture<T> repeat(final Supplier<T> action, final Predicate<T> until) {
    final CompletableFuture<T> futureAction = supplyAsync(action);
    final boolean isMatchFound = futureAction.thenApplyAsync(until::test).join();

    final T suppliedValue = getSuppliedValue(futureAction);

    if (isMatchFound) {
        LOGGER.info("Got a match for value {}.", suppliedValue);

        return futureAction;
    }

    return repeat(() -> suppliedValue, until);
}

private static <T> T getSuppliedValue(final CompletableFuture<T> futureAction) {
    try {
        return futureAction.get();
    } catch (InterruptedException | ExecutionException e) {
        LOGGER.error(e.getMessage());
    }

    return null;
}

测试用例：

@Test
public void testRepeat() {
    Supplier<PasswordAuthentication> action = () -> {
        final String[] passwordVault = new String[] { "password", "secret",
            "secretPassword" };

        final int len = passwordVault.length;

        return new PasswordAuthentication("mickeyMouse",
            passwordVault[ThreadLocalRandom.current().nextInt(len)].toCharArray());
    };
    @SuppressWarnings("static-access")
    Predicate<PasswordAuthentication> until = passwordAuth -> {
        try {
            currentThread().sleep(1000);
        } catch (InterruptedException e) {
            fail(e.getMessage());
        }

        final String password = String.valueOf(passwordAuth.getPassword());

        LOGGER.info("Received password: {}.", password);

        return password.equals("secret");
    };

    repeat(action, until);
}

运行一次，看看选择相同的密码有多奇怪：

2015-01-09 15:41:33.350 [线程 1] [信息] n.a.j.j.c.PracticeQuestionsCh6Test - 收到密码： 秘密密码。 2015-01-09 15:41:34.371 [线程 3] [信息] n.a.j.j.c.PracticeQuestionsCh6Test - 收到密码： 秘密密码。 2015-01-09 15:41:35.442 [线程 5] [信息] n.a.j.j.c.PracticeQuestionsCh6Test - 收到密码： 秘密密码。 2015-01-09 15:41:36.443 [线程 7] [信息] n.a.j.j.c.PracticeQuestionsCh6Test - 收到密码： 秘密密码。 2015-01-09 15:41:37.451 [线程 9] [信息] n.a.j.j.c.PracticeQuestionsCh6Test - 收到密码： 密码。

Answer 1

我发现了错误。我创建的供应商每次都返回相同的值。以下是更正后的代码。但是有一个问题：为什么编译器会强制转换？

public static <T> CompletableFuture<T> repeat(final Supplier<T> action,
        final Predicate<T> until) {
    final CompletableFuture<T> futureAction = supplyAsync(action);

    @SuppressWarnings("unchecked")
    CompletableFuture<T> future = (CompletableFuture<T>) futureAction
        .thenApplyAsync(until::test).thenApply(
            isMatchFound -> isMatchFound ? futureAction : repeat(action, until));

    future.join();

    return future;

}

Answer 2

我认为你正在走一条不必要的复杂道路。由于供应-测试-供应-测试-供应-测试序列是连续的，初始supplyAsync 就是您所需要的。没有理由在异步中进行异步。

这是一个简单的实现：

public static <T> CompletableFuture<T> repeat(final Supplier<T> action,
        final Predicate<T> until) {

    return CompletableFuture.supplyAsync(() -> 
        Stream.generate(action)
                .filter(until)
                .findFirst()
                .get()
    );
}