"Parallel.For" für Java?

MLaw Lösung ist sehr praktisch, Parallel.ForEach. Ich habe ein bisschen eine Veränderung, eine Parallele.Für.

public class Parallel
{
static final int iCPU = Runtime.getRuntime().availableProcessors();

public static <T> void ForEach(Iterable <T> parameters,
                   final LoopBody<T> loopBody)
{
    ExecutorService executor = Executors.newFixedThreadPool(iCPU);
    List<Future<?>> futures  = new LinkedList<Future<?>>();

    for (final T param : parameters)
    {
        Future<?> future = executor.submit(new Runnable()
        {
            public void run() { loopBody.run(param); }
        });

        futures.add(future);
    }

    for (Future<?> f : futures)
    {
        try   { f.get(); }
        catch (InterruptedException e) { } 
        catch (ExecutionException   e) { }         
    }

    executor.shutdown();     
}

public static void For(int start,
                   int stop,
               final LoopBody<Integer> loopBody)
{
    ExecutorService executor = Executors.newFixedThreadPool(iCPU);
    List<Future<?>> futures  = new LinkedList<Future<?>>();

    for (int i=start; i<stop; i++)
    {
        final Integer k = i;
        Future<?> future = executor.submit(new Runnable()
        {
            public void run() { loopBody.run(k); }
        });     
        futures.add(future);
    }

    for (Future<?> f : futures)
    {
        try   { f.get(); }
        catch (InterruptedException e) { } 
        catch (ExecutionException   e) { }         
    }

    executor.shutdown();     
}
}

public interface LoopBody <T>
{
    void run(T i);
}

public class ParallelTest
{
int k;  

public ParallelTest()
{
    k = 0;
    Parallel.For(0, 10, new LoopBody <Integer>()
    {
        public void run(Integer i)
        {
            k += i;
            System.out.println(i);          
        }
    });
    System.out.println("Sum = "+ k);
}

public static void main(String [] argv)
{
    ParallelTest test = new ParallelTest();
}
}

InformationsquelleAutor der Antwort Weimin Xiao

Fork-join-framework in Java 7 ist für die Unterstützung von Parallelität. Aber ich weiß nicht über eine genaue Entsprechung für Parallel.For.

InformationsquelleAutor der Antwort Emil

Eine einfachere option wäre

//A thread pool which runs for the life of the application.
private static final ExecutorService EXEC = 
    Executors.newFixedThreadPool(SOME_NUM_OF_THREADS); 

//later 
EXEC.invokeAll(tasks); //you can optionally specify a timeout.

InformationsquelleAutor der Antwort Peter Lawrey

Synchronisation oft tötet der speedup der parallelen for-Schleifen. Daher die parallele for-Schleifen Häufig müssen Sie Ihre privaten Daten und eine Reduktion des Mechanismus zu verringern, alle threads, die privaten Daten zu umfassen, die ein einzelnes Ergebnis.

So, ich habe erweitert die Parallele.Für die version von Weimin Xiao durch eine Reduktion Mechanismus.

public class Parallel {
public static interface IntLoopBody {
    void run(int i);
}

public static interface LoopBody<T> {
    void run(T i);
}

public static interface RedDataCreator<T> {
    T run();
}

public static interface RedLoopBody<T> {
    void run(int i, T data);
}

public static interface Reducer<T> {
    void run(T returnData, T addData);
}

private static class ReductionData<T> {
    Future<?> future;
    T data;
}

static final int nCPU = Runtime.getRuntime().availableProcessors();

public static <T> void ForEach(Iterable <T> parameters, final LoopBody<T> loopBody) {
    ExecutorService executor = Executors.newFixedThreadPool(nCPU);
    List<Future<?>> futures  = new LinkedList<Future<?>>();

    for (final T param : parameters) {
        futures.add(executor.submit(() -> loopBody.run(param) ));
    }

    for (Future<?> f : futures) {
        try { 
            f.get();
        } catch (InterruptedException | ExecutionException e) { 
            System.out.println(e); 
        }
    }
    executor.shutdown();     
}

public static void For(int start, int stop, final IntLoopBody loopBody) {
    final int chunkSize = (stop - start + nCPU - 1)/nCPU;
    final int loops = (stop - start + chunkSize - 1)/chunkSize;
    ExecutorService executor = Executors.newFixedThreadPool(loops);
    List<Future<?>> futures  = new LinkedList<Future<?>>();

    for (int i=start; i < stop; ) {
        final int iStart = i;
        i += chunkSize;
        final int iStop = (i < stop) ? i : stop;

        futures.add(executor.submit(() -> {
            for (int j = iStart; j < iStop; j++) 
                loopBody.run(j);
        }));     
    }

    for (Future<?> f : futures) {
        try { 
            f.get();
        } catch (InterruptedException | ExecutionException e) { 
            System.out.println(e); 
        }
    }
    executor.shutdown();     
}

public static <T> void For(int start, int stop, T result, final RedDataCreator<T> creator, final RedLoopBody<T> loopBody, final Reducer<T> reducer) {
    final int chunkSize = (stop - start + nCPU - 1)/nCPU;
    final int loops = (stop - start + chunkSize - 1)/chunkSize;
    ExecutorService executor = Executors.newFixedThreadPool(loops);
    List<ReductionData<T>> redData  = new LinkedList<ReductionData<T>>();

    for (int i = start; i < stop; ) {
        final int iStart = i;
        i += chunkSize;
        final int iStop = (i < stop) ? i : stop;
        final ReductionData<T> rd = new ReductionData<T>();

        rd.data = creator.run();
        rd.future = executor.submit(() -> {
            for (int j = iStart; j < iStop; j++) {
                loopBody.run(j, rd.data);
            }
        });
        redData.add(rd);
    }

    for (ReductionData<T> rd : redData) {
        try { 
            rd.future.get();
            if (rd.data != null) {
                reducer.run(result, rd.data);
            }
        } catch (InterruptedException | ExecutionException e) { 
            e.printStackTrace();
        }
    }
    executor.shutdown();     
}
}

Hier ist ein einfacher test Beispiel: eine parallele Zeichen-Zähler unter Verwendung einer nicht-synchronisierten anzeigen.

import java.util.*;

public class ParallelTest {
static class Counter {
    int cnt;

    Counter() {
        cnt = 1;
    }
}

public static void main(String[] args) {
    String text = "More formally, if this map contains a mapping from a key k to a " + 
            "value v such that key compares equal to k according to the map's ordering, then " +
            "this method returns v; otherwise it returns null.";
    Map<Character, Counter> charCounter1 = new TreeMap<Character, Counter>();
    Map<Character, Counter> charCounter2 = new TreeMap<Character, Counter>();

    //first sequentially
    for(int i=0; i < text.length(); i++) {
        char c = text.charAt(i);
        Counter cnt = charCounter1.get(c);
        if (cnt == null) {
            charCounter1.put(c, new Counter());
        } else {
            cnt.cnt++;
        }
    }
    for(Map.Entry<Character, Counter> entry: charCounter1.entrySet()) {
        System.out.println(entry.getKey() + ": " + entry.getValue().cnt);
    }

    //now parallel without synchronization
    Parallel.For(0, text.length(), charCounter2,
        //Creator
        () -> new TreeMap<Character, Counter>(), 
        //Loop Body
        (i, map) -> {
            char c = text.charAt(i);
            Counter cnt = map.get(c);
            if (cnt == null) {
                map.put(c, new Counter());
            } else {
                cnt.cnt++;
            }
        }, 
        //Reducer
        (result, map) -> {
            for(Map.Entry<Character, Counter> entry: map.entrySet()) {
                Counter cntR = result.get(entry.getKey());
                if (cntR == null) {
                    result.put(entry.getKey(), entry.getValue());
                } else {
                    cntR.cnt += entry.getValue().cnt;
                }
            }
        }
    );

    //compare results
    assert charCounter1.size() == charCounter2.size() : "wrong size: " + charCounter1.size() + ", " + charCounter2.size();
    Iterator<Map.Entry<Character, Counter>> it2 = charCounter2.entrySet().iterator();
    for(Map.Entry<Character, Counter> entry: charCounter1.entrySet()) {
        Map.Entry<Character, Counter> entry2 = it2.next();
        assert entry.getKey() == entry2.getKey() && entry.getValue().cnt == entry2.getValue().cnt : "wrong content";
    }

    System.out.println("Well done!");
}
}

InformationsquelleAutor der Antwort Chris

Source code :
https://github.com/crammeur/MyInvoices/blob/master/crammeurLibrairy/src/main/java/ca/qc/bergeron/marcantoine/crammeur/librairy/utils/Parallel.java

[UPDATE]

Parallel-Klasse :

private static final int NUM_CORES = Runtime.getRuntime().availableProcessors();
private static final int MAX_THREAD = NUM_CORES*2;

@NotNull
public static <T> void For(final Collection<T> elements, final Operation<T> operation) {
    ExecutorService executor = Executors.newFixedThreadPool(MAX_THREAD);
    final Iterator<T> iterator = elements.iterator();
    final Runnable runnable = new Runnable() {
        final Callable<Void> callable = new Callable<Void>() {
            @Override
            public Void call() throws Exception {
                T result;
                synchronized (iterator) {
                    result = iterator.next();
                }
                operation.perform(result);
                return null;
            }
        };
        @Override
        public void run() {
            while (iterator.hasNext()) {
                try {
                    synchronized (operation) {
                        if (operation.follow()) {
                            synchronized (callable) {
                                callable.call();
                            }
                            if (!operation.follow()) {
                                break;
                            }
                        } else {
                            break;
                        }
                    }
                } catch (NoSuchElementException e) {
                    break;
                } catch (Exception e) {
                    e.printStackTrace();
                    throw new RuntimeException(e);
                }
            }
        }
    };
    for (int threadIndex=0; threadIndex<MAX_THREAD; threadIndex++) {
        executor.execute(runnable);
    }
    executor.shutdown();
    while (!executor.isTerminated()) {
        try {
            Thread.sleep(0,1);
        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
}

public interface Operation<T,R> {
    R perform(T pParameter);
    boolean follow();
}

javaparallelemultithreading

InformationsquelleAutor der Antwort