Java - многопоточная одна большая петля

Question

Это, наверное, довольно простой вопрос, но поскольку я никогда не работал с потоками, прежде чем я понял, что лучше спросить, а не пытаться найти оптимальное решение самостоятельно.

У меня есть гигантский цикл for, который работает буквально в миллиарды раз. При каждом запуске цикла в соответствии с текущим index программа вычисляет конечный результат в виде числа. Меня интересует только хранение верхнего result (или верхних х результатов) и его соответствующего индекса.

Мой вопрос прост, каков был бы правильный способ запуска этого цикла в потоках, чтобы он использовал все доступные процессоры/ядра.

int topResultIndex; 
double topResult = 0; 

for (i=1; i < 1000000000; ++i) { 
    double result = // some complicated calculation based on the current index 
    if (result > topResult) { 
     topResult = result; 
     topResultIndex = i; 
    } 
} 

Расчет полностью независим для каждого индекса, ресурсы не используются. topResultIndex и topResult, очевидно, будут доступны каждому потоку.

* Обновление: Оба решения Giulio и rolfl являются хорошими, также очень похожими. Могло принять только один из них в качестве моего ответа.

Answer 1

Давайте предположим, что результат вычисляется с помощью calculateResult(long) метода, который является приватным и статическим, а не доступ к любому статическому полю, (он также может быть не статичным, но все же она должна быть поточно-и одновременно исполняемый, надеюсь, ограниченный поток).

Тогда, я думаю, что это будет делать грязную работу:

public static class Response { 
    int index; 
    double result; 
} 

private static class MyTask implements Callable<Response> { 
    private long from; 
    private long to; 

    public MyTask(long fromIndexInclusive, long toIndexExclusive) { 
     this.from = fromIndexInclusive; 
     this.to = toIndexExclusive; 
    } 

    public Response call() { 
     int topResultIndex; 
     double topResult = 0; 

     for (long i = from; i < to; ++i) { 
      double result = calculateResult(i); 
      if (result > topResult) { 
       topResult = result; 
       topResultIndex = i; 
      } 
     } 

     Response res = new Response(); 
     res.index = topResultIndex; 
     res.result = topResult; 
     return res; 
    } 
}; 

private static calculateResult(long index) { ... } 

public Response interfaceMethod() { 
    //You might want to make this static/shared/global 
    ExecutorService svc = Executors.newCachedThreadPool(); 

    int chunks = Runtime.getRuntime().availableProcessors(); 
    long iterations = 1000000000; 
    MyTask[] tasks = new MyTask[chunks]; 
    for (int i = 0; i < chunks; ++i) { 
     //You'd better cast to double and round here 
     tasks[i] = new MyTask(iterations/chunks * i, iterations/chunks * (i + 1)); 
    } 

    List<Future<Response>> resp = svc.invokeAll(Arrays.asList(tasks)); 
    Iterator<Future<Response>> respIt = resp.iterator(); 

    //You'll have to handle exceptions here 
    Response bestResponse = respIt.next().get(); 

    while (respIt.hasNext()) { 
     Response r = respIt.next().get(); 
     if (r.result > bestResponse.result) { 
      bestResponse = r; 
     } 
    } 

    return bestResponse; 
} 

Из моего опыта, это деление на куски намного быстрее, что, имея задачу для каждого индекса (особенно, если расчетная нагрузка для каждого отдельного индекса мал, как будто это возможно. Небольшой я имею в виду менее половины секунды). Это немного сложнее для кода, потому что вам нужно сделать двухэтапную максимизацию (сначала на уровне блоков, затем на глобальном уровне). При этом, если вычисление чисто cpu-based (не слишком сильно нажимает ram), вы должны получить ускорение, почти равное 80% числа физических ядер.

Answer 2

Самый простой способ - использовать ExecutorService и представить свои задачи как Runnable или Callable. Вы можете использовать Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()) для создания ExeuctorService, который будет использовать то же количество потоков, что и процессоры.

Answer 3

Помимо наблюдения за тем, что программа C с OpenMP или некоторыми другими параллельными вычислительными расширениями будет лучшей идеей, способ Java это сделать, чтобы создать «Будущую» задачу, которая вычисляет подмножество проблемы:

private static final class Result { 
    final int index; 
    final double result; 
    public Result (int index, double result) { 
     this.result = result; 
     this.index = index; 
    } 
} 

// Calculate 10,000 values in each thead 
int steps = 10000; 
int cpucount = Runtime.getRuntime().availableProcessors(); 
ExecutorService service = Executors.newFixedThreadPool(cpucount); 
ArrayList<Future<Result>> results = new ArrayList<>(); 
for (int i = 0; i < 1000000000; i+= steps) { 
    final int from = i; 
    final int to = from + steps; 
    results.add(service.submit(new Callable<Result>() { 
     public Result call() { 
       int topResultIndex = -1; 
       double topResult = 0; 
       for (int j = from; j < to; j++) { 
        // do complicated things with 'j' 
         double result = // some complicated calculation based on the current index 
         if (result > topResult) { 
          topResult = result; 
          topResultIndex = j; 
         } 
       } 
       return new Result(topResultIndex, topResult); 
     } 
    }); 
} 

service.shutdown(); 
while (!service.isTerminated()) { 
    System.out.println("Waiting for threads to complete"); 
    service.awaitTermination(10, TimeUnit.SECONDS); 
} 
Result best = null; 
for (Future<Result> fut : results) { 
    if (best == null || fut.result > best.result) { 
     best = fut; 
    } 
} 

System.out.printf("Best result is %f at index %d\n", best.result, best.index); 

Future<Result>