5

我有一些很容易并行化的工作，我想使用 Java 线程在我的四核机器上拆分工作。这是一种应用于旅行商问题的遗传算法。这听起来不容易并行化，但第一个循环很容易并行化。第二部分我谈论实际演变可能是也可能不是，但我想知道我是否因为我实现线程的方式而变慢，或者它是否是算法本身。

此外，如果有人对我应该如何实施我正在尝试做的事情有更好的想法，那将非常感激。

在 main() 中，我有这个：

 final ArrayBlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(numThreads*numIter);
 ThreadPoolExecutor tpool = new ThreadPoolExecutor(numThreads, numThreads, 10, TimeUnit.SECONDS, queue);
 barrier = new CyclicBarrier(numThreads);
 k.init(tpool);

我有一个在 init() 内部完成的循环，如下所示：

for (int i = 0; i < numCities; i++) {
    x[i] = rand.nextInt(width);
    y[i] = rand.nextInt(height);
}

我改成这样：

int errorCities = 0, stepCities = 0;
stepCities = numCities/numThreads;
errorCities = numCities - stepCities*numThreads;

// Split up work, assign to threads                                                                        
for (int i = 1; i <= numThreads; i++) {
    int startCities = (i-1)*stepCities;
    int endCities = startCities + stepCities;

    // This is a bit messy...                                                                              
    if(i <= numThreads) endCities += errorCities;
    tpool.execute(new citySetupThread(startCities, endCities));
}

这是 citySetupThread() 类：

public class citySetupThread implements Runnable {
    int start, end;

    public citySetupThread(int s, int e) {
        start = s;
        end = e;
    }
    public void run() {
        for (int j = start; j < end; j++) {
            x[j] = ThreadLocalRandom.current().nextInt(0, width);
            y[j] = ThreadLocalRandom.current().nextInt(0, height);
        }

        try {
            barrier.await();
        } catch (InterruptedException ie) {
            return;
        } catch (BrokenBarrierException bbe) {
            return;
        }
    }
}

上面的代码在程序中运行一次，所以它是我的线程构造的一个测试用例（这是我第一次使用 Java 线程）。我在一个真正的关键部分实现了同样的事情，特别是遗传算法的进化部分，其类如下：

public class evolveThread implements Runnable {
    int start, end;

    public evolveThread(int s, int e) {
        start = s;
        end = e;
    }
    public void run() {
        // Get midpoint                                                                                            
        int n = population.length/2, m;

        for (m = start; m > end; m--) {
            int i, j;
            i = ThreadLocalRandom.current().nextInt(0, n);

            do {
                j = ThreadLocalRandom.current().nextInt(0, n);
            } while(i == j);

            population[m].crossover(population[i], population[j]);
            population[m].mutate(numCities);
        }

        try {
            barrier.await();
        } catch (InterruptedException ie) {
            return;
        } catch (BrokenBarrierException bbe) {
            return;
        }

    }
}

它存在于在 init() 中调用的函数 Evolution() 中，如下所示：

for (int p = 0; p < numIter; p++) evolve(p, tpool);

是的，我知道这不是非常好的设计，但出于其他原因，我坚持使用它。进化内部是相关部分，如下所示：

// Threaded inner loop                                                                                     
int startEvolve = popSize - 1,
endEvolve = (popSize - 1) - (popSize - 1)/numThreads;

// Split up work, assign to threads                                                                        
for (int i = 0; i < numThreads; i++) {
    endEvolve = (popSize - 1) - (popSize - 1)*(i + 1)/numThreads + 1;
    tpool.execute(new evolveThread(startEvolve, endEvolve));
    startEvolve = endEvolve;
}

// Wait for our comrades                                                                                   
try {
     barrier.await();
} catch (InterruptedException ie) {
     return;
} catch (BrokenBarrierException bbe) {
     return;
}

population[1].crossover(population[0], population[1]);
population[1].mutate(numCities);
population[0].mutate(numCities);

// Pick out the strongest                                                                                      
Arrays.sort(population, population[0]);
current = population[0];
generation++;

我真正想知道的是：

  • “队列”有什么作用？我是否可以为池中的所有线程执行的尽可能多的作业创建队列？如果大小不够大，我会得到 RejectedExecutionException。我只是决定做 numThreads*numIterations 因为那会有多少工作（对于我之前提到的实际进化方法）。不过这很奇怪..如果barrier.await（）正在工作，我不应该这样做，这导致我......

  • 我是否正确使用了 barrier.await()？目前我在两个地方都有它：在 Runnable 对象的 run() 方法中，以及在执行所有作业的 for 循环之后。我原以为只需要一个，但如果我删除一个或另一个，我会得到错误。

  • 我怀疑线程的争用，因为这是我可以从荒谬的减速中收集到的唯一东西（它确实与输入参数成比例）。我想知道这是否与我如何实现线程池和障碍有关。如果没有，那么我想我将不得不查看 crossover() 和 mutate() 方法。

4

2 回答 2

4

首先，我认为您打算如何使用 CyclicBarrier 可能存在错误。目前，您正在使用执行程序线程的数量作为参与方的数量对其进行初始化。但是，您还有另一方；主线程。所以我认为你需要这样做：

barrier = new CyclicBarrier(numThreads + 1);

我认为这应该可行，但我个人认为这是对屏障的奇怪使用。

当使用工作队列线程池模型时，我发现使用 Semaphore 或 Java 的 Future 模型更容易。

对于信号量：

class MyRunnable implements Runnable {
  private final Semaphore sem;

  public MyRunnable(Semaphore sem) {
    this.sem = sem;
  }

  public void run() {
    // do work

    // signal complete
    sem.release()
  }
}

然后在你的主线程中：

Semaphore sem = new Semaphore(0);

for (int i = 0; i < numJobs; ++i) {
  threadPool.execute(new MyRunnable(sem));
}

sem.acquire(numJobs);

它确实与屏障做同样的事情，但我发现更容易考虑工作任务“发出信号”它们已完成，而不是再次与主线程“同步”。

例如，如果您查看CyclicBarrier JavaDoc中的示例代码，则调用barrier.await()是在 worker 内部的循环内。所以它实际上是在同步多个长时间运行的工作线程，而主线程没有参与屏障。barrier.await()在循环外的工作人员结束时调用更多的是信号完成。

于 2012-12-02T01:25:16.470 回答
1

当您增加任务数量时，您会增加使用每个任务添加的开销。这意味着您希望尽量减少任务数量，即与您拥有的 CPU 数量相同。对于某些任务，当工作负载不均匀时，使用双倍 CPU 数量可能会更好。

BTW：您不需要在每个任务中设置障碍，您可以通过调用每个任务来等待每个任务的未来完成get()。

于 2012-12-02T11:28:23.703 回答