Java 意外的并发结果

Question

在测试并发时，我发现了一些意想不到的事情。

使用 concurrentHashMap 和 AtomicLong 控制并发。

public class HumanRepository {

    private final static Map<Long, Human> STORE = new ConcurrentHashMap<>();
    private AtomicLong sequence = new AtomicLong();

    public void save(Human human) {
        STORE.put(sequence.incrementAndGet(), human);
    }

    public int size() {
        return STORE.size();
    }


    public Long getSeq() {
        return sequence.get();
    }
}

我测试了多线程保存。

    @Test
    void name() throws NoSuchMethodException, InterruptedException {
        final int threads = 3_500;
        final ExecutorService es = Executors.newFixedThreadPool(threads);
        final CountDownLatch count = new CountDownLatch(threads);
        final HumanRepository repository = new HumanRepository();

        for (int i = 0; i < threads; i++) {
            try {
                es.execute(() -> repository.save(new Human("aa")));
            } finally {
                count.countDown();
            }
        }

        count.await();

        System.out.println("seq = " + repository.getSeq());
        System.out.println("size = " + repository.size());
    }

我同时测试了3500个线程。我期望的结果是 seq 和 size 都是 3500。

但有时我会得到 seq=3499，size=3500。那真是怪了。奇怪的是seq出来的不是3500，虽然size是3500，seq是3499也没意义

不知道为什么map里面的data number和seq不一样，3500出不来

**如果你在count.await();之后做Thread.sleep(400L);，奇怪的是，seq的值为3500

Answer 1

您实际上并不是在等待所有任务完成。也就是说，如果你得到3500/3500输出，那是偶然的。

具体来说，您在安排作业后减少主线程上的倒计时闩锁，而不是作业内部，一旦完成.这意味着您的 countdownlatch 基本上只是另一个不进行任何 inter-thread 通信的美化循环变量。尝试这样的事情：

    for (int i = 0; i < threads; i++) {
        es.execute(() -> {
          repository.save(new Human("aa"));
          count.countDown();
        });
    }

Answer 2

您在执行 HumanRepository.save() 的线程外调用 count.countDown()。所以可能主线程没有同步完成线程。

因此您可能会看到 repository.getSeq() 的结果，而一个线程是运行。您可以尝试使用以下代码吗？

        final int threads = 3_500;
        final ExecutorService es = Executors.newFixedThreadPool(threads);
        final CountDownLatch count = new CountDownLatch(threads);
        final HumanRepository repository = new HumanRepository();

        for (int i = 0; i < threads; i++) {
            try {
                es.execute(() -> {
                    repository.save(new Human("aa"));
                    count.countDown();
                });
            } finally {
                
            }
        }

        count.await();

        System.out.println("seq = " + repository.getSeq());
        System.out.println("size = " + repository.size());

Java 意外的并发结果

Java unexpected concurrent result

java

concurrency