该CompletionStage的Javadoc指出:
[...]如果一个阶段的计算突然以(未经检查的)异常或错误终止,那么所有需要完成的依赖阶段也会异常完成,并且CompletionException将异常作为其原因.
看作异常完成总是包含异常,CompletionException为什么这样做exceptionally(),whenComplete()并handle()代表异常Throwable代替CompletionException？
这很重要,因为它可以防止人们直接在这些方法中重新抛出异常.
这些方法是否有可能接收除以外的异常CompletionException？或者我可以安全地强制演员到这种类型吗？
(我在本地运行了一些测试,以及挖掘CompletableFuture源代码,乍一看,我没有看到如何抛出任何其他类型的异常.)
我有几种CompletionStage方法可以链接.问题是第一个的结果将决定是否应该执行下一个.现在,实现这一目标的唯一方法似乎是将"特殊"参数传递给next,CompletionStage因此它不会执行完整的代码.例如:
public enum SomeResult {
RESULT_1,
RESULT_2,
RESULT_3
}
public CompletionStage<SomeResult> someMethod(SomeArgument someArgument) {
return CompletableFuture.supplyAsync(() -> {
// loooooong operation
if (someCondition)
return validValue;
else
return null;
}).thenCompose(result -> {
if (result != null)
return someMethodThatReturnsACompletionStage(result);
else
return CompletableFuture.completedFuture(null);
}).thenApply(result -> {
if (result == null)
return ChainingResult.RESULT_1;
else if (result.someCondition())
return ChainingResult.RESULT_2;
else
return ChainingResult.RESULT_3;
});
}
因为整个代码依赖于第一个代码someCondition(如果它是false结果将是RESULT_1,如果不是那么整个代码应该被执行)这个结构对我来说看起来有点难看.有没有办法决定是否应该执行2nd(thenCompose(...))和3rd(thenApply(...))方法？
我对这段代码有疑问:
@Async
public CompletableFuture<String> doFoo() {
CompletableFuture<String> fooFuture = new CompletableFuture<>();
try {
String fooResult = longOp();
fooFuture.complete(fooResult);
} catch (Exception e) {
fooFuture.completeExceptionally(e);
}
return fooFuture;
}
问题是:doFoo仅在longOp完成后(正确或异常)返回fooFuture,因此返回已经完成的期货,或者Spring在执行主体之前做了一些魔法并返回？如果代码在longOp()上被阻塞,你会如何表达计算被送到执行程序？
也许这个？还有其他方法吗？
@Async
public CompletableFuture<String> doFoo() {
CompletableFuture<String> completableFuture = new CompletableFuture<>();
CompletableFuture.runAsync(() -> {
try {
String fooResult = longOp();
completableFuture.complete(fooResult);
} catch (Exception e) {
completableFuture.completeExceptionally(e);
}
});
return completableFuture;
}
我最近刚刚开始使用 CompletableFuture，但遇到了一个问题，我有 N 个请求待办事项。
每个请求应发送到 2 个不同的端点，并应比较 JSON 格式的结果。由于我有大量的请求待办事项，并且我不知道每个请求需要多长时间，因此我想限制等待结果的时间，例如 3 秒左右。
所以我写了这个测试代码：
public class MainTest {
private static final Logger logger = LoggerFactory.getLogger(MainTest.class);
private Instant start;
public static void main(String[] args) {
MainTest main = new MainTest();
main.start();
}
public void start(){
String req1 = "http://localhost:8080/testing";
String req2 = "http://127.0.0.1:8095/testing2";
ExecutorService exec = Executors.newCachedThreadPool();
start = Instant.now();
CompletableFuture<String> comp1 = CompletableFuture.supplyAsync(() -> doReq(req1), exec);
CompletableFuture<String> comp2 = CompletableFuture.supplyAsync(() -> doReq(req2), exec);
List<CompletableFuture<String>> completables = List.of(comp1,comp2);
logger.info("Waiting completables");
CompletableFuture<List<String>> a = …我在学习时有一个问题CompletableFuture。将get()/join()方法阻塞调用。如果我不打电话给他们中的任何一个怎么办？
此代码调用get()：
// Case 1 - Use get()
CompletableFuture.runAsync(() -> {
try {
Thread.sleep(1_000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Hello");
}).get();
System.out.println("World!");
Thread.sleep(5_000L); // Don't finish the main thread
输出：
Hello
World!
此代码既不调用get()也不调用join()：
// Case 2 - Don't use get()
CompletableFuture.runAsync(() -> {
try {
Thread.sleep(1_000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Hello");
});
System.out.println("World!");
Thread.sleep(5_000L); // For don't finish main thread
输出：
World!
Hello …CompletableFuture::supplyAsync(() -> IO bound queries)
如何为CompletableFuture :: supplyAsync选择Executor以避免污染ForkJoinPool.commonPool().
有许多选项Executors(newCachedThreadPool,newWorkStealingPool,newFixedThreadPool等)
我在这里阅读了关于新ForkJoinPool的内容
如何为我的用例选择合适的？
java executorservice java-8 threadpoolexecutor completable-future
我想要一个仅表示完成的CompletableFuture(例如,我没有返回值).
我可以将CompletableFuture实例化为:
CompletableFuture<Void> future = new CompletableFuture<> ();
但是我应该为完整的方法提供什么？例如,我做不到
future.complete(new Void());
在java-9 中,引入了类中的新方法completeOnTimeoutCompletableFuture:
public CompletableFuture<T> completeOnTimeout(T value, long timeout,
TimeUnit unit) {
if (unit == null)
throw new NullPointerException();
if (result == null)
whenComplete(new Canceller(Delayer.delay(
new DelayedCompleter<T>(this, value),
timeout, unit)));
return this;
}
我不明白为什么它在其实现中使用静态 ScheduledThreadPoolExecutor:
static ScheduledFuture<?> delay(Runnable command, long delay,
TimeUnit unit) {
return delayer.schedule(command, delay, unit);
}
哪里
static final ScheduledThreadPoolExecutor delayer;
static {
(delayer = new ScheduledThreadPoolExecutor(
1, new DaemonThreadFactory())).
setRemoveOnCancelPolicy(true);
}
对我来说这是一种非常奇怪的方法,因为它可能成为整个应用程序的瓶颈:唯一一个ScheduledThreadPoolExecutor只有一个线程保留在池中以执行所有可能的CompletableFuture任务？
我在这里错过了什么？
PS它看起来像:
1)这段代码的作者不愿意提取这种逻辑,而是倾向于重用ScheduledThreadPoolExecutor …
java multithreading threadpoolexecutor java-9 completable-future
我解决了一个非常具体的问题,其解决方案似乎是基本的:
我的(Spring)应用程序的类加载器层次结构是这样的: SystemClassLoader -> PlatformClassLoader -> AppClassLoader
如果我使用Java CompleteableFuture来运行线程.该ContextClassLoader线程的是: SystemClassLoader -> PlatformClassLoader -> ThreadClassLoader
因此,AppClassLoader虽然我必须访问任何类,但我无法访问任何类,因为所有外部库类都驻留在那里.
源代码库非常大,所以我不希望/不能将所有与线程相关的部分重写为其他内容(例如,将自定义执行程序传递给每个调用).
所以我的问题是:我怎样才能创建线程,例如CompleteableFuture.supplyAsync()使用AppClassLoader父母作为父母？(而不是PlatformClassloader)
我发现ForkJoinPool用于创建线程.但在我看来,一切都是静态的和最终的.所以我怀疑即使在系统属性中设置自定义ForkJoinWorkerThreadFactory也会有所帮助.或者是吗？
编辑以回答评论中的问题:
你在哪里部署？这是在jetty/tomcat /任何JEE容器内运行吗？
你有什么确切的问题？
您提交给supplyAsync()的作业是从AppClassLoader创建的,不是吗？
在supplyAsync从被称为MainThread它使用AppClassLoader.但是,调试应用程序会显示所有此类线程都具有PlatformClassLoader父级.至于我的理解,这是因为ForkJoinPool.commonPool()是在应用程序启动期间构建的(因为它是静态的),所以使用默认的类加载器作为父类PlatformClassLoader.因此,此池中的所有线程都将PlatformClassLoader作为ContextClassLoader的父级(而不是AppClassLoader).
当我在内部创建自己的执行程序MainThread并将此执行程序传递给supplyAsync所有工作时 - 我可以在调试期间看到确实现在AppClassLoader是我的父项ThreadClassLoader.这似乎证实了我在第一种情况下的假设,即公共池MainThread至少不是在它自己使用时创建的 …
java ×8
java-8 ×3
spring ×2
asynchronous ×1
classloader ×1
concurrency ×1
java-9 ×1
task ×1
unirest-java ×1