Volatile 没有按预期工作

Question

所以我正在阅读 Brian Goetz 的 JCIP 并编写了以下代码来试验 volatile 行为。

public class StatefulObject {

    private static final int NUMBER_OF_THREADS = 10;

    private volatile State state;

    public StatefulObject() {
        state = new State();
    }

    public State getState() {
        return state;
    }

    public void setState(State state) {
        this.state = state;
    }

    public static class State {
        private volatile AtomicInteger counter;

        public State() {
            counter = new AtomicInteger();
        }

        public AtomicInteger getCounter() {
            return counter;
        }

        public void setCounter(AtomicInteger counter) {
            this.counter = counter;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        StatefulObject object = new StatefulObject();

        ExecutorService executorService = Executors.newFixedThreadPool(NUMBER_OF_THREADS);

        AtomicInteger oldCounter = new AtomicInteger();
        AtomicInteger newCounter = new AtomicInteger();

        object.getState().setCounter(oldCounter);

        ConcurrentMap<Integer, Long> lastSeen = new ConcurrentHashMap<>();
        ConcurrentMap<Integer, Long> firstSeen = new ConcurrentHashMap<>();
        lastSeen.put(oldCounter.hashCode(), 0L);
        firstSeen.put(newCounter.hashCode(), Long.MAX_VALUE);

        List<Future> futures = IntStream.range(0, NUMBER_OF_THREADS)
            .mapToObj(num -> executorService.submit(() -> {
                for (int i = 0; i < 1000; i++) {
                    object.getState().getCounter().incrementAndGet();
                    lastSeen.computeIfPresent(object.getState().getCounter().hashCode(), (key, oldValue) -> Math.max(oldValue, System.nanoTime()));
                    firstSeen.computeIfPresent(object.getState().getCounter().hashCode(), (key, oldValue) -> Math.min(oldValue, System.nanoTime()));
                }
            })).collect(Collectors.toList());

        executorService.shutdown();

        object.getState().setCounter(newCounter);

        futures.forEach(future -> {
            try {
                future.get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        });

        System.out.printf("Counter: %s\n", object.getState().getCounter().get());
        long lastSeenOld = lastSeen.get(oldCounter.hashCode());
        long firstSeenNew = firstSeen.get(newCounter.hashCode());
        System.out.printf("Last seen old counter: %s\n", lastSeenOld);
        System.out.printf("First seen new counter: %s\n", firstSeenNew);
        System.out.printf("Old was seen after the new: %s\n", lastSeenOld > firstSeenNew);
        System.out.printf("Old was seen %s nanoseconds after the new\n", lastSeenOld - firstSeenNew);
    }
}

所以我希望 newCounter 总是在最后一次看到 oldCounter 之后才第一次看到（我希望所有线程都注意到更新，所以 none 正在引用陈旧的计数器).为了观察这种行为，我使用了两张地图。但令人惊讶的是，我经常得到这样的输出：

Counter: 9917
Last seen old counter: 695372684800871
First seen new counter: 695372684441226
Old was seen after the update: true
Old was seen 359645 nanoseconds after the new

你能解释一下我哪里错了吗？

提前致谢！

Answer 1

您看到的不是 volatile 的影响，而是同步对 ConcurrentMap<> lastSeen 的影响。

让我们假设所有十个线程几乎同时启动。每个人几乎并行执行 object.getState().getCounter().incrementAndGet();，从而增加 oldCounter.

接下来，这些线程尝试执行 lastSeen.computeIfPresent(object.getState().getCounter().hashCode(), (key, oldValue) -> Math.max(oldValue, System.nanoTime()));。这意味着，它们都并行计算 object.getState().getCounter().hashCode()，每个都获得 oldCounter 的相同哈希码，然后使用相同的哈希值调用 ConcurrentHashMap.computeIfPresent(Integer, ..)。

由于所有这些线程都尝试更新同一个键的值，ConcurrentHashMap 必须同步这些更新 - 仅。

在第一个线程更新 lastSeen 期间，主线程执行 object.getState().setCounter(newCounter);，因此第一个线程将为 newCounter 执行 firstSeen，而多个线程还在等更新lastSeen.

为了获得更好的结果，最好将信息收集步骤与分析步骤分开。

例如，线程可以将计数器哈希码和更新时间戳捕获到您在完成所有计算后分析的数组中。

Answer 2

您的观察背后的原因不是 java 中的错误；）而是您的代码中有一个错误。在您的代码中，您不能保证对 lastseen 和 firstSeen 映射的 computeIfPresent 调用是原子执行的（请参阅 Javadocs，computeIfPresent 不是原子的）。这意味着在您获得 object.getState().getCounter() 和实际更新地图之间存在时间间隔。

如果设置 newCounter 发生在线程 A 在此间隙中（在获取纳米时间之前但已经获取计数器引用 - 旧）并且线程 B 恰好在获取 object.getState().getCounter() 之前。因此，如果更新了这个确切的时刻计数器引用，线程 A 将更新旧的计数器键，而线程 B 将更新新的。如果线程 B 在线程 A 之前花费了纳米时间（这可能发生，因为这些是分开的线程，我们无法知道实际的 cpu 调度是什么），这可能会完美地导致您的观察。

我想我的解释很清楚了。还有一件事需要澄清，在 State class 中，您也已将 AtomicInteger counter 声明为易变的。这不是必需的，因为 AtomicInteger 本质上是易变的。没有 "non-volatile" Atomic** s.

我只是更改了您的代码中的一些内容以忽略上述问题：

import java.util.Collections;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class StatefulObject {

    private static final int NUMBER_OF_THREADS = 10;

    private volatile State state;

    public StatefulObject() {
        state = new State();
    }

    public State getState() {
        return state;
    }

    public void setState(State state) {
        this.state = state;
    }

    public static class State {
        private volatile AtomicInteger counter;

        public State() {
            counter = new AtomicInteger();
        }

        public AtomicInteger getCounter() {
            return counter;
        }

        public void setCounter(AtomicInteger counter) {
            this.counter = counter;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        StatefulObject object = new StatefulObject();

        ExecutorService executorService = Executors.newFixedThreadPool(NUMBER_OF_THREADS);

        AtomicInteger oldCounter = new AtomicInteger();
        AtomicInteger newCounter = new AtomicInteger();

        object.getState().setCounter(oldCounter);

        List<Long> oldList = new CopyOnWriteArrayList<>();
        List<Long> newList = new CopyOnWriteArrayList<>();

        List<Future> futures = IntStream.range(0, NUMBER_OF_THREADS)
            .mapToObj(num -> executorService.submit(() -> {
                for (int i = 0; i < 1000; i++) {
                    long l = System.nanoTime();
                    object.getState().getCounter().incrementAndGet();
                    if (object.getState().getCounter().equals(oldCounter)) {
                        oldList.add(l);
                    } else {
                        newList.add(l);
                    }
                }
            })).collect(Collectors.toList());

        executorService.shutdown();

        object.getState().setCounter(newCounter);

        futures.forEach(future -> {
            try {
                future.get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        });

        System.out.printf("Counter: %s\n", object.getState().getCounter().get());
        Collections.sort(oldList);
        Collections.sort(newList);
        long lastSeenOld = oldList.get(oldList.size() - 1);
        long firstSeenNew = newList.get(0);
        System.out.printf("Last seen old counter: %s\n", lastSeenOld);
        System.out.printf("First seen new counter: %s\n", firstSeenNew);
        System.out.printf("Old was seen after the new: %s\n", lastSeenOld > firstSeenNew);
        System.out.printf("Old was seen %s nanoseconds after the new\n", lastSeenOld - firstSeenNew);
    }
}

Volatile 没有按预期工作

Volatile doesn't work as expected

java

concurrency

multithreading

volatile

thread-safety