Java 如何将 Callback 转 Promise？

Question

Java 如何将 Callback 转 Promise？

场景

之前 Java 使用回调函数实现了等待指定资源加载然后执行回调，现在想修改为 Promise 风格的，有什么好的方法么？

例如

void watch(Predicate<String> f, Consumer<Stat> callback)

转换成

Future<Stat> watch(Predicate<String> f)

顺便说一下，吾辈在 JavaScript 中倒是习惯了 Promise 的异步，但 Java 中却是不怎么熟悉呢。。。
下面是将 setTimeout 封装为 Promise 的代码

var wait = ms => new Promise(res => setTimeout(res, ms))

所以，有什么好的解决方案么？

---

吾辈知道 Java 中有 CompletedFuture 和 CompletableFuture，然而不知道的是如何创建一个 CompletableFuture 对象并在构造过程中使用绑定的回调函数

例如吾辈确实可以将下面这种代码转换为柯里化的函数

default void watch(String path, Consumer<Stat> callback) {
    watch(map -> map.get(ListUtil.ListDiffState.right).stream()
            .filter(stat -> Objects.equals(stat.getPath(), path))
            .findFirst(), callback);
}

转换后

default Consumer<Consumer<Stat>> watch(String path) {
    return callback -> watch(path, callback);
}

然而，吾辈却无法将之转换为 CompletedFuture 异步风格（拿不到 CompletedFuture 绑定的回调函数），求教！

Answer 1

你需要第三方库
https://projectreactor.io
https://github.com/ReactiveX/...

Answer 2

已经解决问题了，也因此写了一篇博客 使用 Java 实现 setTimeout/setInterval，内容如下：

使用 Java 实现 setTimeout/setInterval

场景

之前想把 Java 代码中使用回调函数的方法改成 Promise 风格，苦思冥想而不得其解。然而突发奇想之下，吾辈尝试在 Java 中实现 JavaScript 的 setTimeout/setInterval，并在之后想到了如何封装回调为 Promise，所以便先在此将这个想法的写出来以供参考。

Promise 是 ES6 添加的一个重要的元素，它将回调函数压平为了一级调用，并在 ES7 的 async/await 中彻底改变了异步的使用方式！

实现

public class AsyncUtil {
    private static final Logger log = LoggerFactory.getLogger(AsyncUtil.class);

    /**
     * 将当前线程休眠指定的时间
     *
     * @param millis 毫秒
     */
    public static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 实现 JavaScript 中的 setTimeout
     * 注：由于 {@link CompletableFuture#cancel(boolean)} 方法的限制，该定时器无法强制取消
     *
     * @param ms 等待时间
     * @return 异步对象
     */

    public static CompletableFuture<Void> setTimeout(long ms) {
        return CompletableFuture.runAsync(() -> sleep(ms));
    }

    /**
     * 实现等待指定资源加载完成
     * 注：由于 {@link CompletableFuture#cancel(boolean)} 方法的限制，该定时器无法强制取消
     *
     * @param condition 临界条件
     * @return 异步对象
     */
    public static CompletableFuture<Void> waitResource(Supplier<Boolean> condition) {
        return CompletableFuture.runAsync(() -> {
            while (!condition.get()) {
                sleep(100);
            }
        });
    }

    /**
     * 实现 JavaScript 中的 setInterval 周期函数
     * 该方法并不是非常精确的定时器，仅适用于一般场景，如有需要请使用 {@link ScheduledExecutorService} 类
     * 注：由于 {@link CompletableFuture#cancel(boolean)} 方法的限制，该定时器无法强制取消
     *
     * @param ms       间隔毫秒数
     * @param runnable 回调函数
     * @return 永远不会完成的异步对象
     */
    public static CompletableFuture<Void> setInterval(long ms, Runnable runnable) {
        return CompletableFuture.runAsync(() -> {
            //noinspection InfiniteLoopStatement
            while (true) {
                try {
                    runnable.run();
                    sleep(ms);
                } catch (Exception e) {
                    log.error("使用 setInterval 发生异常: ", e);
                }
            }
        });
    }
}

使用

public class AsyncUtilTest {
    private final Logger log = LoggerFactory.getLogger(getClass());

    public static void main(String[] args) {
        final AsyncUtilTest asyncUtilTest = new AsyncUtilTest();
        asyncUtilTest.setTimeout();
        asyncUtilTest.waitResource();
        asyncUtilTest.setInterval();
        AsyncUtil.sleep(4000);
    }

    @Test
    public void setTimeout() {
        log.info("setTimeout completed before time: {}", LocalDateTime.now());
        AsyncUtil.setTimeout(1000)
                .thenRunAsync(() -> log.info("setTimeout completed after time: {}", LocalDateTime.now()));
    }

    @Test
    public void waitResource() {
        log.info("waitResource completed before time: {}", LocalDateTime.now());
        final AtomicInteger i = new AtomicInteger(1);
        AsyncUtil.waitResource(() -> i.get() == 3)
                .thenRunAsync(() -> log.info("waitResource completed after time: {}", LocalDateTime.now()));
        AsyncUtil.sleep(2);
        i.set(3);
    }

    @Test
    public void setInterval() {
        log.info("setInterval completed before time: {}", LocalDateTime.now());
        final CompletableFuture<Void> future = AsyncUtil.setInterval(100, () -> log.info("setInterval in the loop, current time: {}", LocalDateTime.now()));
        AsyncUtil.sleep(500);
        future.complete(null);
        AsyncUtil.sleep(1000);
    }
}