Java 용 "Parallel.For"?

Java 용 "Parallel.For"?

---

Java 용 .net 버전과 동등한 Parallel.For 가 있는지 궁금합니다 .

누군가가 있다면 예를 들어주세요. 감사!

---

가장 가까운 것은 다음과 같습니다.

ExecutorService exec = Executors.newFixedThreadPool(SOME_NUM_OF_THREADS);
try {
    for (final Object o : list) {
        exec.submit(new Runnable() {
            @Override
            public void run() {
                // do stuff with o.
            }
        });
    }
} finally {
    exec.shutdown();
}

TheLQ의 의견에 따라 SUM_NUM_THREADS를 다음과 같이 설정합니다. Runtime.getRuntime().availableProcessors();

편집 : 기본 "Parallel.For"구현을 추가하기로 결정했습니다.

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

    private static final ExecutorService forPool = Executors.newFixedThreadPool(NUM_CORES * 2, new NamedThreadFactory("Parallel.For"));

    public static <T> void For(final Iterable<T> elements, final Operation<T> operation) {
        try {
            // invokeAll blocks for us until all submitted tasks in the call complete
            forPool.invokeAll(createCallables(elements, operation));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static <T> Collection<Callable<Void>> createCallables(final Iterable<T> elements, final Operation<T> operation) {
        List<Callable<Void>> callables = new LinkedList<Callable<Void>>();
        for (final T elem : elements) {
            callables.add(new Callable<Void>() {
                @Override
                public Void call() {
                    operation.perform(elem);
                    return null;
                }
            });
        }

        return callables;
    }

    public static interface Operation<T> {
        public void perform(T pParameter);
    }
}

Parallel.For 사용 예

// Collection of items to process in parallel
Collection<Integer> elems = new LinkedList<Integer>();
for (int i = 0; i < 40; ++i) {
    elems.add(i);
}
Parallel.For(elems, 
 // The operation to perform with each item
 new Parallel.Operation<Integer>() {
    public void perform(Integer param) {
        System.out.println(param);
    };
});

이 구현은 Parallel.ForEach 와 실제로 더 유사하다고 생각합니다.

편집 누군가가 관심이 있다면 GitHub에 올려 놓았습니다. GitHub의 Parallel For

---

MLaw의 솔루션은 매우 실용적인 Parallel.ForEach입니다. Parallel.For를 만들기 위해 약간의 수정을 추가했습니다.

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();
}
}

---

mlaw 제안에 따라 CountDownLatch를 추가합니다. submit ()을 줄이기 위해 chunksize를 추가하십시오.

4 백만 개의 항목 배열로 테스트했을 때이 제품은 Core i7 2630QM CPU에서 순차적 for ()보다 5 배 빠른 속도를 제공합니다.

public class Loop {
    public interface Each {
        void run(int i);
    }

    private static final int CPUs = Runtime.getRuntime().availableProcessors();

    public static void withIndex(int start, int stop, final Each body) {
        int chunksize = (stop - start + CPUs - 1) / CPUs;
        int loops = (stop - start + chunksize - 1) / chunksize;
        ExecutorService executor = Executors.newFixedThreadPool(CPUs);
        final CountDownLatch latch = new CountDownLatch(loops);
        for (int i=start; i<stop;) {
            final int lo = i;
            i += chunksize;
            final int hi = (i<stop) ? i : stop;
            executor.submit(new Runnable() {
                public void run() {
                    for (int i=lo; i<hi; i++)
                        body.run(i);
                    latch.countDown();
                }
            });
        }
        try {
            latch.await();
        } catch (InterruptedException e) {}
        executor.shutdown();
    }

    public static void main(String [] argv) {
        Loop.withIndex(0, 9, new Loop.Each() {
            public void run(int i) {
                System.out.println(i*10);
            }
        });
    }
}

---

Java 7의 포크 조인 프레임 워크는 동시성 지원을위한 것입니다. 그러나 나는 정확히 Parallel.For.

---

이 주제에 대한 저의 기여는 https://github.com/pablormier/parallel-loops 입니다. 사용법은 매우 간단합니다.

Collection<String> upperCaseWords = 
    Parallel.ForEach(words, new Parallel.F<String, String>() {
        public String apply(String s) {
            return s.toUpperCase();
        }
    });

스레드 수 (기본적으로 캐시 된 스레드 풀을 사용)와 같은 일부 동작 측면을 변경할 수도 있습니다.

Collection<String> upperCaseWords = 
            new Parallel.ForEach<String, String>(words)
                .withFixedThreads(4)
                .apply(new Parallel.F<String, String>() {
                    public String apply(String s) {
                        return s.toUpperCase();
                    }
                }).values();

모든 코드는 하나의 Java 클래스에 자체 포함되며 JDK보다 더 많은 종속성이 없습니다. 또한 Java 8을 사용하여 기능적 방식으로 병렬화하는 새로운 방법 을 확인 하는 것이 좋습니다.

---

더 간단한 옵션은

// 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.

---

Parallel.For와 동등한 것이 Java 확장으로 사용 가능합니다. Ateji PX라고 불리며, 함께 플레이 할 수있는 무료 버전이 있습니다. http://www.ateji.com/px/index.html

parallel.for와 정확히 동일하며 비슷하게 보입니다.

For ||

wikipedia의 더 많은 예제 및 설명 : http://en.wikipedia.org/wiki/Ateji_PX

Java IMO에서 닫힌 것

---

동기화는 종종 병렬 for 루프의 속도 향상을 방해합니다. 따라서 병렬 for 루프는 종종 개인 데이터와 단일 결과를 구성하도록 모든 스레드 개인 데이터를 줄이기위한 감소 메커니즘이 필요합니다.

그래서 저는 Weimin Xiao감소 메커니즘 에 의해 Parallel.For 버전을 확장했습니다 .

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();     
}
}

다음은 간단한 테스트 예입니다. 동기화되지 않은 맵을 사용하는 병렬 문자 카운터입니다.

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!");
}
}

---

Parallel.For, Parallel.ForEach, Parallel.Tasks 및 분할 된 병렬 루프를 수행 할 수있는 업데이트 된 Java Parallel 클래스가 있습니다. 소스 코드는 다음과 같습니다.

이러한 병렬 루프를 사용하는 예는 다음과 같습니다.

public static void main(String [] argv)
{
    //sample data
    final ArrayList<String> ss = new ArrayList<String>();

    String [] s = {"a", "b", "c", "d", "e", "f", "g"};
    for (String z : s) ss.add(z);
    int m = ss.size();

    //parallel-for loop
    System.out.println("Parallel.For loop:");
    Parallel.For(0, m, new LoopBody<Integer>()
    {
        public void run(Integer i)
        {
           System.out.println(i +"\t"+ ss.get(i));   
        }       
    });   

   //parallel for-each loop
   System.out.println("Parallel.ForEach loop:");
   Parallel.ForEach(ss, new LoopBody<String>()
   {
       public void run(String p)
       {
           System.out.println(p);               
       }       
   });

   //partitioned parallel loop
   System.out.println("Partitioned Parallel loop:");
   Parallel.ForEach(Parallel.create(0, m), new LoopBody<Partition>()
   {
       public void run(Partition p)
       {
           for(int i=p.start; i<p.end; i++)
               System.out.println(i +"\t"+ ss.get(i));
       }
   });

   //parallel tasks
   System.out.println("Parallel Tasks:");
   Parallel.Tasks(new Task []
   {
       //task-1
       new Task() {public void run()
       {
           for(int i=0; i<3; i++)
               System.out.println(i +"\t"+ ss.get(i));
       }},

       //task-2
       new Task() {public void run()
       {
           for (int i=3; i<6; i++)
               System.out.println(i +"\t"+ ss.get(i));
       }}   
   });
}

---

내 경우에는 ForkJoinPool과 IntStream이 매우 유용하다는 것을 알았습니다 (제한된 수의 스레드가있는 병렬 For).

씨#:

static void MathParallel(int threads)
        {
            Parallel.For(1, partitions, new ParallelOptions { MaxDegreeOfParallelism = threads }, (i) => {
                partitionScores[i] = Math.Sin(3*i);
            });
        }

및 Java 동등 :

static void mathParallel(int threads) {
        ForkJoinPool pool = new ForkJoinPool(threads);
            pool.submit(()-> IntStream.range(0, partitions).parallel().forEach(i -> {
                partitionScores[i] = Math.sin(3*i);
            }));
        pool.shutdown();
        while (!pool.isTerminated()){
        }
    }

---

이것이 내가 Java 7 이하에서 사용하는 것입니다.

Java 8의 경우 forEach ()를 사용할 수 있습니다.

[업데이트]

병렬 클래스 :

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

public static <T2 extends T, T> void For(final Iterable<T2> elements, final Operation<T> operation) {
    if (elements != null) {
        final Iterator<T2> iterator = elements.iterator();
        if (iterator.hasNext()) {
            final Throwable[] throwable = new Throwable[1];
            final Callable<Void> callable = new Callable<Void>() {
                boolean first = true;
                @Override
                public final Void call() throws Exception {
                    if ((first || operation.follow()) && iterator.hasNext()) {
                        T result;
                        result = iterator.next();
                        operation.perform(result);
                        if (first) {
                            synchronized (this) {
                                first = false;
                            }
                        }
                    }
                    return null;
                }
            };
            final Runnable runnable = new Runnable() {
                @Override
                public final void run() {
                    while (iterator.hasNext()) {
                        try {
                            synchronized (callable) {
                                callable.call();
                            }
                            if (!operation.follow()) {
                                break;
                            }
                        } catch (Throwable t) {
                            t.printStackTrace();
                            synchronized (throwable) {
                                throwable[0] = t;
                            }
                            throw new RuntimeException(t);
                        }
                    }
                }
            };
            final ExecutorService executor = Executors.newFixedThreadPool(MAX_THREAD);
            for (int threadIndex=0; threadIndex<MAX_THREAD && iterator.hasNext(); threadIndex++) {
                executor.execute(runnable);
            }
            executor.shutdown();
            while (!executor.isTerminated()) {
                try {
                    Thread.sleep(0,1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    throw new RuntimeException(e);
                }
            }
            if (throwable[0] != null) throw new RuntimeException(throwable[0]);
        }
    }
}

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

예

@Test
public void test() {
    List<Long> longList = new ArrayList<Long>();
    for (long i = 0; i < 1000000; i++) {
        longList.add(i);
    }
    final List<Integer> integerList = new LinkedList<>();
    Parallel.For((Iterable<? extends Number>) longList, new Parallel.Operation<Number>() {

        @Override
        public void perform(Number pParameter) {
            System.out.println(pParameter);
            integerList.add(pParameter.intValue());
        }

        @Override
        public boolean follow() {
            return true;
        }
    });
    for (Number num : integerList) {
        System.out.println(num);
    }
}

자바 병렬 멀티 스레딩

참고 URL : https://stackoverflow.com/questions/4010185/parallel-for-for-java