Java - 多线程一个大循环

Question

这可能是一个非常简单的问题，但由于我从未使用过线程，因此我认为最好是问而不是试图完全自己找到最佳解决方案。

我有一个巨大的 for 循环，它运行了数十亿次。在每次on循环运行时，程序根据当前的index，计算出一个数字形式的最终结果。我只对存储前result（或前x 个结果）及其相应索引感兴趣。

我的问题很简单，在线程中运行这个循环的正确方法是什么，以便它使用所有可用的 CPU/内核。

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

最简单的方法是使用ExecutorService 并将您的任务提交为Runnable 或Callable。您可以使用Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()) 创建一个ExeuctorService，它将使用与处理器数量相同的线程数。

Answer 2

除了观察到带有 OpenMP 或其他一些并行计算扩展的 C 程序会是一个更好的主意之外，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>

Answer 3

假设结果是由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（不会过多地推动 ram），您应该获得几乎等于物理内核数量 80% 的加速。