在连续的 java CompletionStages 之间对共享上下文对象所做的最新更改对于执行 lambda 的每个线程是不是始终可见

Posted 2023-02-25

【中文标题】在连续的 java CompletionStages 之间对共享上下文对象所做的最新更改对于执行 lambda 的每个线程是不是始终可见

【英文标题】：Are latest changes made to shared context object between consecutive java CompletionStages always visible to each thread executing the lambda在连续的 java CompletionStages 之间对共享上下文对象所做的最新更改对于执行 lambda 的每个线程是否始终可见

【发布时间】：2022-01-02 23:53:40

【问题描述】：

package org.***.example;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MeetPlay 

    private static final class Context 
        List<String> data = new ArrayList<>();
    

    public static void main(String[] args) 

        ExecutorService executor = Executors.newFixedThreadPool(
                Runtime.getRuntime().availableProcessors());

        Context context = new Context();
        CompletableFuture.completedFuture(context)
                .thenAcceptAsync(c -> context.data.add("1"), executor)
                .thenAcceptAsync(__ -> context.data.add("2"), executor)
                .thenAcceptAsync(__ -> 
                    if (context.data.contains("1") && context.data.contains("2")) 
                        System.out.println("Will that always be the case");
                     else 
                        System.out.println("Or not");
                    
                , executor);
    

我目前的理解是，无论执行程序使用多少线程、阶段和项目，或者上下文对象有多“复杂”（例如，有更多嵌套字段），由于 java 完成阶段之间的保证发生之前发生。

工作中的一位同事认为这不能保证，并说他已经通过经验证明了这一点，但是，我并没有亲眼看到它。在那之前，我想知道你们怎么想，最重要的是WHY。参考将不胜感激！

编辑：问题是关于共享上下文的内存可见性保证，而不是每个阶段是否在前一个阶段完成后执行。

【参考方案1】：

你是对的 - 这些阶段是按顺序执行的。这可以通过运行以下代码来证明：

    CompletableFuture.completedFuture(context)
        .thenAcceptAsync(c -> 
            context.data.add("1");
            System.out.println("1");
            try 
                Thread.sleep(1000);
             catch (InterruptedException e) 
            , executor)
        .thenAcceptAsync(__ -> 
            context.data.add("2");
            System.out.println("2");
            try 
                Thread.sleep(1000);
             catch (InterruptedException e) 
        , executor)
        .thenAcceptAsync(__ -> 
            if (context.data.contains("1") && context.data.contains("2")) 
                System.out.println("Will that always be the case");
             else 
                System.out.println("Or not");
            
        , executor);

结果将是：

1
(pause 1 sec)
2
(pause 1 sec)
Will that always be the case

发生这种情况是因为从 thenAcceptAsync 返回的 CompletionStage 仅在操作完成后完成。 IE。如果动作异常完成，CompletionStage 也会异常完成。此行为与 thenAccept 相同。 thenAcceptAsync 中的“Async”仅表示该操作将使用另一个执行器执行。

如果您想要并行执行，请考虑以下代码：

    CompletableFuture<Context> f = CompletableFuture.completedFuture(context);
    f.thenAcceptAsync(c -> 
        try 
            Thread.sleep(1000);
         catch (InterruptedException e) 
        context.data.add("1");
        System.out.println("1");
    , executor);
    f.thenAcceptAsync(__ -> 
        try 
            Thread.sleep(1000);
         catch (InterruptedException e) 
        context.data.add("2");
        System.out.println("2");
    , executor);
    f.thenAcceptAsync(__ -> 
        if (context.data.contains("1") && context.data.contains("2")) 
            System.out.println("Will that always be the case");
         else 
            System.out.println("Or not");
        
    , executor);

这里 3 个阶段添加到同一个阶段，而不是一个接一个。一个阶段的完成和另一个阶段的开始之间没有依赖关系。所以阶段可以并行执行。结果将是：

Or not
(pause 1 sec)
2
1

【讨论】：

嘿，我编辑了我最初的问题，并澄清了实际问题是什么。我知道阶段将按顺序调用。

@festiv 我相信顺序执行意味着在内存模型方面发生之前。 CompletionStage 的文档没有明确说明，但它在内部使用 Executor 来保证发生前发生。这个答案也可能有帮助***.com/questions/34427972/…