没有 Lambda 的 CompletableFuture.supplyAsync()

问题描述

我正在努力处理 supplier<U> 等的函数式风格并创建可测试的代码。

所以我有一个 InputStream 被分成异步处理的块，我想知道它们什么时候完成。为了编写可测试的代码，我将处理逻辑外包给它自己的 Runnable：

public class StreamProcessor {
    
    public CompletableFuture<Void> process(InputStream in) {
        List<CompletableFuture> futures = new ArrayList<>();
        while (true) {
            try (SizeLimitInputStream chunkStream = new SizeLimitInputStream(in,100)) {
                byte[] data = IoUtils.toByteArray(chunkStream);
                CompletableFuture<Void> f = CompletableFuture.runAsync(createTask(data));
                futures.add(f);
                
            } catch (EOFException ex) {
                // end of stream reached
                break;
            } catch (IOException ex) {
                return CompletableFuture.FailedFuture(ex);
            }
        }
        return CompletableFuture.allOf(futures.toArray(CompletableFuture<?>[]::new));
    }
    
    ChunkTask createTask(byte[] data) {
        return new ChunkTask(data);
    }
    
    public class ChunkTask implements Runnable {
        final byte[] data;

        ChunkTask(byte[] data) {
            this.data = data;
        }

        @Override
        public void run() {
            try {
                // do something
            } catch (Exception ex) {
                // checked exceptions must be wrapped
                throw new RuntimeException(ex);
            }
        }
        
    }
}

这很有效，但会带来两个问题：

处理代码不能返回任何内容；毕竟是 Runnable。
在 ChunkTask.run() 中捕获的任何已检查异常都必须包装到 RuntimeException 中。解开失败的组合 CompletableFuture 返回 RuntimeException，需要再次解开以达到原始原因 - 与 IOException 相反。

所以我正在寻找一种使用 CompletableFuture.supplyAsync() 来做到这一点的方法，但我无法弄清楚如何在没有 lambdas 的情况下做到这一点（不好测试）或从处理逻辑。

解决方法

我可以想到两种方法：

1.使用supplyAsync：

使用 CompletableFuture.supplyAsync 时，您需要一个供应商而不是一个可运行的：

    public static class ChunkTask implements Supplier<Object> {
        final byte[] data;

        ChunkTask(byte[] data) {
            this.data = data;
        }

        @Override
        public Object get() {
            Object result = ...;
            // Do something or throw an exception
            return result;
        }
    }

然后：

CompletableFuture
    .supplyAsync( new ChunkTask( data ) )
    .whenComplete( (result,throwable) -> ... );

如果在 Supplier.get() 中发生异常，它会被传播，您可以在 CompletableFuture.whenComplete、CompletableFuture.handle 或 CompletableFuture.exceptionally 中看到它。

2.将 CompletableFuture 传递给线程

您可以将 CompletableFuture 传递给 ChunkTask：

    public class ChunkTask implements Runnable {
        final byte[] data;
        private final CompletableFuture<Object> future;

        ChunkTask(byte[] data,CompletableFuture<Object> future) {
            this.data = data;
            this.future = future;
        }

        @Override
        public void run() {
            try {
                Object result = null;
                // do something
                future.complete( result );
            } catch (Throwable ex) {
                future.completeExceptionally( ex );
            }
        }
    }

那么逻辑就变成了：

while (true) {
    CompletableFuture<Object> f = new CompletableFuture<>();
    try (SizeLimitInputStream chunkStream = new SizeLimitInputStream(in,100)) {
        byte[] data = IOUtils.toByteArray(chunkStream);
        startThread(new ChunkTask(data,f));
        futures.add(f);
     } catch (EOFException ex) {
         // end of stream reached
         break;
     } catch (IOException ex) {
         f.completeExceptionally( ex );
         return f;
     }
}

也许，第 2 条可以让您更灵活地管理异常。

completable-future java java