JavaScript的Promise.all()

Promise是JavaScript异步编程的一种解决方案，在ES6中引入。

通过Promise.all()可以实现对一组异步请求的统一处理，等待所有异步执行完成之后调用回调函数。

其实，这种并发执行同步等待的需求在Java并发编程中也很常见，那么，是否可以通过Java也来实现这样一个Promise类呢？

使用Java实现Promise.all()

使用工具

• CountDownLatch：Java并发工具包中有CountDownLatch类允许一个或多个线程等待其他线程的一系列操作完成。
• ThreadPoolExecutor：通过线程池实现多线程的并发执行

实现

public class Promise {

    private static ExecutorService executorService = Executors.newScheduledThreadPool(16);

    private Promise() {
        throw new AssertionError();
    }

    /**
     * 实现并发同时地对某个action并发执行并返回执行结果
     * 实现思路：
     * 并发创建所有执行的线程，并通过锁(start)阻塞等待着
     * 在创建所有执行的线程后(ready)开始计时，并解锁然所有的线程启动
     * 通过另外一个锁（done）记录执行完的线程
     * 主线程只需关心3点
     * - 所有线程是否准备好
     * - 准备好的话开始计时并解锁开始执行
     * - 等待执行完毕
     *
     * @param callableList 要并发执行的列表
     * @return list 执行结果，list.item为null的话表示执行异常
     * @throws InterruptedException 异常
     */
    public static <T> List<T> all(final List<Callable<T>> callableList) throws InterruptedException {
        final List<T> result = new ArrayList<>();
        int length = callableList.size();
        final CountDownLatch ready = new CountDownLatch(length);
        final CountDownLatch start = new CountDownLatch(1);
        final CountDownLatch done = new CountDownLatch(length);
        for (final Callable<T> callable : callableList) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    ready.countDown();
                    try {
                        start.await();
                        T t = callable.call();
                        result.add(t);
                    } catch (Exception e) {
                        // interrupt when exception
                        Thread.currentThread().interrupt();
                        // set null mean exception
                        result.add(null);
                        e.printStackTrace();
                    } finally {
                        done.countDown();
                    }
                }
            });
        }
        ready.await();
        long startnano = System.nanoTime();
        start.countDown();
        done.await();
        long cause = System.nanoTime() - startnano;
        System.out.println(String.format("Promise all done,cause time millSecond: %s", cause / 1000000));
        return result;
    }

}

效果

• 测试

public void promiseAllTest() throws Exception{

    List<Callable<String>> callables = new ArrayList<>();

    for (int i = 0; i < 10; i++) {
        int finalI = i;
        callables.add(new Callable<String>() {
            @Override
            public String call() throws Exception {
                int millis = new Random().nextInt(10000);
                Thread.sleep(millis);
                System.out.println(String.format("thread%s sleep %s millis" ,finalI,millis));
                return "Thread" + finalI;
            }
        });
    }

    List<String> result = Promise.all(callables);

    System.out.println(result);
    System.out.println("done...");

}

• 测试结果

thread1 sleep 732 millis
thread2 sleep 758 millis
thread7 sleep 976 millis
thread8 sleep 1397 millis
thread5 sleep 1513 millis
thread0 sleep 2221 millis
thread3 sleep 4885 millis
thread6 sleep 5221 millis
thread4 sleep 7101 millis
thread9 sleep 7634 millis
Promise all done,cause time millSecond: 7638
[Thread1, Thread2, Thread7, Thread8, Thread5, Thread0, Thread3, Thread6, Thread4, Thread9]
done...

总结

本文只是通过原生Java实现简单版本的Promise.all()，可用于简单的并发编程，但是对于实际高并发应用还需要优化，如对线程池的优化,还有中断的处理等。

参考

• 《Effective Java》第二版第十章第69条：并发工具优先于wait和notify