java - 执行器服务和可运行

我有一个包含要为每个像素计算的数据的列表(例如，列表大小= 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:
根据沃克斯特的回答，我尝试了以下方法

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块，为什么？

最佳答案

让我感到困惑的是，您的所有可运行对象都使用相同的迭代器。更让我惊讶的是，您在迭代器上迭代时生成了可运行对象，但是这些可运行对象还操纵了迭代器。这段代码可能会(并且将由您的问题证明)会导致一系列竞争状况并因此而引起头痛。

我建议以下内容:

获取相机迭代器

清空缓冲区

从迭代器中读取前x个(例如1000个)样本到缓冲区

中

使用缓冲区创建一个可运行对象，它将对它的1000个条目进行一些工作

将可运行对象提交到服务，然后返回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秒后会击中断点，并且映射包含与字符串“Done”配对的所有Integer。

关于java - 执行器服务和可运行，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/29075100/