一百万次迭代后 Java AtomicInteger 不等于一百万（包括最小示例）

Question

我正在编写一个程序，它计算数组中 0 到 100 万（不包括）之间整数的平方。我最多使用 8 个线程（包括）。
为了索引数组，我使用了一个原子整数。
我希望无论线程数如何，run 方法中的 for 循环体都会执行一百万次。
为了计算它被执行的次数，我使用了另一个原子整数。

   static AtomicInteger cnt = new AtomicInteger(0);
    static AtomicInteger ii = new AtomicInteger(0);
    static long[] arr = new long[1_000_000];

    public static class Worker extends Thread {

        public static void main(String[] args) throws IOException, InterruptedException {
            int maxThreads = Runtime.getRuntime().availableProcessors() * 2;
            for (int n = 1; n <= maxThreads; n++) {
                int numThreads = n;
                Thread[] threads = new Thread[numThreads];
                ii = new AtomicInteger(0);
                for (int i = 0; i < threads.length; i++) {
                    threads[i] = new Worker();
                    threads[i].start();
                }
                for (Thread t : threads) {
                    t.join();
                }
                System.out.printf("%d threads, cnt is %d\n" , threads.length, cnt.get());
                cnt.set(0);
            }
        }

        @Override
        public void run() {
            for (int i = ii.get(); i < arr.length; i = ii.getAndIncrement()) {
                arr[i] = (long)i*i;
                cnt.getAndIncrement();
            }
        }
    }

执行的预期结果是：

1 threads, cnt is 1000000
2 threads, cnt is 1000000
3 threads, cnt is 1000000
4 threads, cnt is 1000000
5 threads, cnt is 1000000
6 threads, cnt is 1000000
7 threads, cnt is 1000000
8 threads, cnt is 1000000

但是在运行时我得到以下信息：

1 threads, cnt is 1000001
2 threads, cnt is 1000002
3 threads, cnt is 1000003
4 threads, cnt is 1000002
5 threads, cnt is 1000003
6 threads, cnt is 1000002
7 threads, cnt is 1000002
8 threads, cnt is 1000005

你能帮我调试吗？

Answer 1

有一些小问题，例如ii = new AtomicInteger(0)运行之间的分配或IOException永远不会发生的抛出声明。虽然分配 toii在这里没有影响，因为它发生在没有线程访问的点ii，它可能会分散注意力，因为它与多线程代码的既定代码模式不同。您应该ii按照cnt在两次运行之间重置的方式进行重置。

实际问题是循环的起点：

for (int i = ii.get(); i < arr.length; i = ii.getAndIncrement()) {

您正在读取 usingget而不增加值，因此多个线程可能会读取相同的值，从而导致随后的数据竞争arr[i] = (long)i*i;（此处未引起注意，但当然应该避免）并执行不必要的迭代次数，正如您注意到的那样到cnt更新。您应该getAndIncrement()像后续迭代一样使用初始索引，以确保每个线程访问不同的数组索引。

static final AtomicInteger cnt = new AtomicInteger(0);
static final AtomicInteger ii = new AtomicInteger(0);
static final long[] arr = new long[1_000_000];

public static class Worker extends Thread {

    public static void main(String[] args) throws InterruptedException {
        int maxThreads = Runtime.getRuntime().availableProcessors() * 2;

        for (int numThreads = 1; numThreads <= maxThreads; numThreads++) {
            Thread[] threads = new Thread[numThreads];
            for (int i = 0; i < threads.length; i++) {
                threads[i] = new Worker();
                threads[i].start();
            }
            for (Thread t : threads) {
                t.join();
            }
            System.out.printf("Used %d threads, cnt is %d\n" , numThreads, cnt.get());
            cnt.set(0);
            ii.set(0);
        }
    }

    @Override
    public void run() {
        for(int i = ii.getAndIncrement(); i < arr.length; i = ii.getAndIncrement()) {
            arr[i] = (long)i*i;
            cnt.getAndIncrement();
        }
    }
}

Used 1 threads, cnt is 1000000
Used 2 threads, cnt is 1000000
Used 3 threads, cnt is 1000000
Used 4 threads, cnt is 1000000
Used 5 threads, cnt is 1000000
Used 6 threads, cnt is 1000000
Used 7 threads, cnt is 1000000
Used 8 threads, cnt is 1000000
Used 9 threads, cnt is 1000000
Used 10 threads, cnt is 1000000
Used 11 threads, cnt is 1000000
Used 12 threads, cnt is 1000000
Used 13 threads, cnt is 1000000
Used 14 threads, cnt is 1000000
Used 15 threads, cnt is 1000000
Used 16 threads, cnt is 1000000

请注意，复杂的并行处理框架不使用这种原子索引更新，而是在启动线程之前根据预期的目标并行度将范围拆分为大小相等的子范围，因此每个线程可以使用普通的本地索引在其预先分配的范围内循环多变的。

您可以在使用Arrays.parallelSetAll(arr, i -> (long)i*i);或 Stream API时免费获得。