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块现在永远不会进入,为什么？

Answer 1

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

我建议如下：

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

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

更新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”配对的所有整数。