java - ExecutorService 和 Runnable

Question

我有一个列表，其中包含要为每个像素计算的数据(例如，列表大小 = 1024x768)。现在我想通过列表进行多线程迭代，并将每个像素的计算保存在 HashMap 中。但无论我做什么，我都无法做到正确。我尝试了几种方法，最后一种是这个:

        ConcurrentMap<T, Color> map = new ConcurrentHashMap<T, Color>();

        ExecutorService pool = Executors.newFixedThreadPool(4);

        Iterator<T> it = camera.iterator();
        while (it.hasNext()) {
            Runnable run = () -> {
                int i = 0;
                while (it.hasNext() && i < 1000) {
                    i++;
                    T cameraRay = it.next();
                    if (object.collide(cameraRay.getRay()) == null)
                        map.put(cameraRay, BG_COLOR);
                    else
                        map.put(cameraRay, this.shader.shade(cameraRay.getRay(), object.collide(cameraRay.getRay())).getColor());
                }
            };
            pool.execute(run);
        }
        pool.shutdown();
        try {
            if (pool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS)) {
                System.out.println("Mapsize: " + map.size());
                // Draw Image:
                map.forEach((ray, color) -> {image.setColor(ray, color);});
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

注意迭代器 hasNext()方法是同步的。问题有时是堆问题，或者只是 HashMap 的大小小于列表大小。

我想我没有理解有关 Runnables 或 ExecutorService 的正确内容。

在此感谢任何帮助。

编辑:我添加了一个 System.out.println(i)就在 i++ 之前陈述。尽管检查了 i < 1000在某个时候突然出现以下内容:

507
169
86624
625
626
Exception in thread "pool-2-thread-2" java.lang.OutOfMemoryError: Java heap space
Exception in thread "pool-2-thread-3" java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at java.lang.reflect.Method.invoke(Unknown Source)
    at com.sun.javafx.application.LauncherImpl.launchApplicationWithArgs(Unknown Source)
    at com.sun.javafx.application.LauncherImpl.launchApplication(Unknown Source)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at java.lang.reflect.Method.invoke(Unknown Source)
    at sun.launcher.LauncherHelper$FXHelper.main(Unknown Source)
Caused by: java.lang.OutOfMemoryError: Java heap space
    at java.util.concurrent.ThreadPoolExecutor.addWorker(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor.execute(Unknown Source)
    at raytracer.impl.ParallelRenderer.render(ParallelRenderer.java:78)
    at raytracer.ImageViewer.main(ImageViewer.java:118)
    ... 11 more
Exception in thread "pool-2-thread-4" java.lang.OutOfMemoryError: Java heap space
java.lang.OutOfMemoryError: Java heap space
    at raytracer.impl.TriangleImpl.collide(TriangleImpl.java:87)
    at raytracer.impl.SimpleScene.collide(SimpleScene.java:27)
    at raytracer.impl.ParallelRenderer.lambda$0(ParallelRenderer.java:71)
    at raytracer.impl.ParallelRenderer$$Lambda$48/24559708.run(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.lang.Thread.run(Unknown Source)

编辑 2:根据Warkst的回答，我尝试了以下

Iterator<T> it = camera.iterator();
List<T> buffer = new ArrayList<T>(1000);
while (it.hasNext()) {
    buffer.add(it.next());
    if (buffer.size() >= 1000 || !it.hasNext()) {
        Runnable run = () -> {
            for (T cameraRay : buffer) {
                if (object.collide(cameraRay.getRay()) == null) // No collision
                    map.put(cameraRay, BG_COLOR);
                else
                    map.put(cameraRay, this.shader.shade(cameraRay.getRay(), object.collide(cameraRay.getRay())).getColor());
            }
        };
        pool.execute(run);
        buffer.clear();
    }
}

但非常奇怪的是，Runnable现在从未输入 block ，为什么？

Answer 1

让我感到困惑的是，您的可运行对象都使用相同的迭代器。更让我惊讶的是，您在遍历迭代器时生成可运行对象，但这些可运行对象还操纵迭代器。这段代码可以(并且将会，正如你的问题所证明的那样)导致一堆竞争条件和随之而来的头痛。

我建议如下:

1. 获取相机迭代器
2. 制作一个空缓冲区
3. 将迭代器中的前 x 个(例如 1000 个)样本读入缓冲区
4. 使用缓冲区创建一个可运行对象，它将对其 1000 个条目执行一些操作
5. 将 runnable 提交给服务并返回到 2。重复直到迭代器不再有下一个。

假设您对数据的处理比在相机上迭代一次(明显)慢，这应该可以解决问题。如果不是这种情况，就真的没有理由使用多线程。

更新 2

我已将我的代码示例更新为有效的东西:

public static void main(String[] args) throws InterruptedException {
    ConcurrentHashMap<Integer, String> map = new ConcurrentHashMap<>();
    ExecutorService pool = Executors.newFixedThreadPool(4);
    Iterator<Integer> it = getIt();
    Task t = new Task(map);
    while (it.hasNext()) {
        t.add(it.next());
        if (t.size()>=1000 || !it.hasNext()) {
            pool.submit(t);
            t = new Task(map);
        }
    }
    pool.shutdown();
    pool.awaitTermination(1, TimeUnit.DAYS);

    // Breakpoint here to inspect map
    System.out.println("Done!");
}

与

private static Iterator<Integer> getIt(){
    return new Iterator<Integer>() {

        private int nr = 0;

        @Override
        public boolean hasNext() {
            return nr < 20000;
        }

        @Override
        public Integer next() {
            return nr++;
        }
    };
}

和

private static class Task extends ArrayList<Integer> implements Runnable{
    private final ConcurrentHashMap<Integer, String> map;

    public Task(ConcurrentHashMap<Integer, String> map) {
        this.map = map;
    }

    @Override
    public void run() {
        try{
            for (Integer i : this) {
                // Simulate some crunching: write to the stdout in 10 iterations for each number: 10 000 prints for each Runnable
                for (int j = 0; j < 10; j++) {
                    System.out.println("Iteration "+j+" for "+i);
                }
                // Store something in the map, namely that this Integer, or T in your case, has been processed
                map.put(i, "Done");
            }
        } catch(Exception e){
            e.printStackTrace();
        }
    }
}

断点在大约 20-30 秒后命中， map 包含与字符串“Done”配对的所有整数。