java - 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 之后才第一次看到(我希望所有线程都注意到更新，所以没有一个线程引用过时的计数器)。为了观察这种行为，我使用了两张 map 。但令人惊讶的是，我不断得到这样的输出:

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

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

如果设置newCounter当线程 A 处于此间隙(在获取纳秒时间之前，但已经获取计数器引用 - 旧)和线程 B 恰好在获取 object.getState().getCounter() 之前发生。 。因此，如果此时计数器引用已更新，线程 A 将更新旧的计数器 key ，而线程 B 将更新新的计数器 key 。如果线程 B 在线程 A 之前花费了纳秒时间(这种情况可能会发生，因为这些是单独的线程，我们无法知道实际的 cpu 调度是什么)，这可能会完全导致您的观察结果。

我想我的解释已经很清楚了。还有一件事需要澄清，在 State类，您已声明 AtomicInteger counter也同样不稳定。这是不需要的，因为 AtomicInteger 本质上是 volatile 的。不存在“非 volatile ”Atomic** 。

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

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);
    }
}