Cách tốt nhất để tổng hợp đồng thời

Question

Tôi đang cố gắng tính toán một số lượng lớn. Để tăng tốc độ tính toán, tôi muốn sử dụng đa luồng. Mỗi chủ đề cần tính toán một số và cuối cùng là một tổng được tính toán.

Tôi đã từng nhìn thấy một cái gì đó mà làm việc với một SumThread và Collector trông như sau:

public BigInteger compute(int p) { 
    Collector c = new Collector(p); 

    for(T element : Collection<T> bigCollection) { 
     new SumThread(c) { 

      @Override 
      protected void doTheJob() { 
       long big = someVeryComplexCalculation(element, ...); //n! 
       receive(BigInteger.valueOf(big)); 
      } 

     } 
    } 

    if(collector.isReady()) 
     return collector.getResult(); 

    return null; 
} 

public class Collector { 

    private int numberOfProcesses; 
    private int numberOfAllowedProcesses; 
    private BigInteger result; 

    public Collector(int n) { 
     numberOfAllowedProcesses = n; 
     numberOfProcesses = 0; 
     result = BigInteger.ZERO; 
    } 

    synchronized public void enter() throws InterruptedException { 
     if (numberOfProcesses == numberOfAllowedProcesses) wait(); 
     numberOfProcesses++; 
    } 

    synchronized public void leave() { 
     numberOfProcesses--; 
     notify(); 
    } 

    synchronized public void register(BigInteger v) { 
     result = result.add(v); 
    } 

    synchronized public boolean isReady() throws InterruptedException { 
     while (numberOfProcesses > 0) wait(); 
     return true; 
    } 

    ... 
} 

public abstract class SumThread extends Thread { 

    private Collector collector; 

    public SumThread(Collector c) throws InterruptedException { 
     collector = c; 
     collector.enter(); 
    } 

    abstract protected void doTheJob(); //complex calculations can be done in here 

    public void receive(BigInteger t) { 
     collector.register(t); 
    } 

    public void run() { 
     doTheJob(); 
     collector.leave(); 
    } 
} 

tôi nghĩ rằng tôi có thể dễ dàng làm tốt hơn điều này bằng cách sử dụng thay vì làm mới Thread s liên tục thích:

public BigInteger compute(int p) { 
    ExecutorService pool = Executors.newFixedThreadPool(p); 
    Future<BigInteger>[] futures = new Future<BigInteger>[bigCollection.size()]; 
    int i = 0; 

    for(T element : Collection<T> bigCollection) { 
     futures[i++] = p.submit(new Callable<BigInteger>() { 

      @Override 
      public BigInteger call() { 
       long big = someVeryComplexCalculation(element, ...); //n! 
       return BigInteger.valueOf(big); 
      } 

     } 
    } 

    // or with ExecutorCompletionService, but the loop remains I guess 
    BigInteger res = BigInteger.ZERO 
    for(Future<BigInteger> f : futures) 
     res = res.add(f.get()); 

    return res; 
} 

Mã này không quản lý tốt hơn giải pháp SumThread - Collector. Tôi cũng đã nhìn thấy những thứ về LongAdder ví dụ, nhưng tôi sẽ cần một số adder cho BigInteger s ...

Câu hỏi của tôi như vậy là: cách tốt nhất để tính toán một khoản đồng thời là gì? Nó là một trong những điều trên hay là có một cách hoàn toàn khác (nhưng tốt hơn)?

Answer 1

Như bạn được đề cập LongAdder được thêm vào trong Java-8 và sử dụng các biến có hiệu quả cuối cùng, tôi giả định rằng bạn đang sử dụng Java-8. Trong phiên bản này là cách tốt nhất để giải quyết nhiệm vụ của bạn là sử dụng Stream API:

BigInteger result = bigCollection.parallelStream() 
        .map(e -> BigInteger.valueOf(someVeryComplexCalculation(e, ...))) 
        .reduce(BigInteger.ZERO, BigInteger::add); 

Vấn đề của bạn là cổ điển bản đồ giảm nhiệm vụ mà bạn nên chuyển đổi mỗi phần tử của một số bộ sưu tập, sau đó kết hợp các kết quả của sự biến đổi cá nhân vào kết quả cuối cùng. API luồng có khả năng song song các tác vụ đó khá hiệu quả mà không cần bất kỳ công việc thủ công nào. Trong Oracle JDK, các tác vụ được thực thi trong common ForkJoinPool pool theo mặc định sẽ tạo ra nhiều luồng như nhiều lõi CPU mà bạn có.

Answer 2

Tou có hai giải pháp:

Trong lần đầu tiên, tôi sẽ đề xuất sử dụng Fork-Join Framework từ JDK7 cho nhiệm vụ này:

• https://docs.oracle.com/javase/tutorial/essential/concurrency/forkjoin.html
• http://www.javacodegeeks.com/2013/02/java-7-forkjoin-framework-example.html

Bạn sẽ cần triển khai RecursiveTask

Là giải pháp thứ hai (JDK8) sẽ sử dụng luồng pararell, giống như @ tagir-valeev đã đề xuất.

Trong cả hai trường hợp, tùy thuộc vào bạn đang sử dụng cái gì và phiên bản Java nào bạn đang sử dụng.