А что тут сложного? только очередь переделать на Queue, обработчики будут спать на ней, вычислители, захватив - будут будить, обработчики обработав кусок - уменьшать переменную обратного отсчета и будить вычислителей.
| Код |
import java.util.Formatter;
import java.util.Queue;
import java.util.LinkedList;
/**
* Главный класс - Обработка одного задания
*/
public class ParallelComputation2 implements Runnable{
final Queue<CompPart> query; // очередь запросов на обработку
int countDown = 0;
private static final int PROCESSORS = Runtime.getRuntime().availableProcessors(); //константа-количество процессоров
public ParallelComputation2(Queue<CompPart> query) {
this.query = query;
}
public void run() {
final int taskNo = Integer.parseInt(Thread.currentThread().getName());
double[][] data = new double[10][10];// данные
for (int i = 0; i < data.length; i++ ) {
double[] row = data[i];
for (int j = 0; j < row.length; j++) {
row[j] = i + j + taskNo;
}
}
synchronized (query) { //захват очереди
countDown = data.length; // переменная обратного отсчета
for (int i = 0; i < data.length; i++) {
// просто пихаем в очередь часть, обработчики выдерут по очереди и обработают
query.add(new CompPart(i, data, this));
}
query.notifyAll(); // будим обработчиков (если дрыхнут в ожидании данных)
}
synchronized (this) { // захват нашего объекта
// может быть ситуация, когда запустив обработчиков, планировщик так долго не давал
// этой нити отработать, что обработчики все сделали. Данная проверка исключает ошибочное ожидание на
//этом объекте в такой ситуации.
while (countDown != 0) { // while - потому что JVM может ошибочно разбудить поток раньше времени.
try {
this.wait(); // ну, а если хотя бы одна часть необработана - дрыхнем, пока нас не разбудят.
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
// выводная форма собирается через StringBuilder внутри класса Formatter, чтобы не смешались
// данные на печати из разных нитей
Formatter f = new Formatter();
f.format("All parts processed, show result thread ").format(Thread.currentThread().getName()).format(" \n");
for (double[] row :data) {
for (double aRow : row) {
f.format("%4.1f ",aRow);
}
f.format("\n");
}
System.out.println(f); // один-единственный синхронный вывод на печать.
// извлекаем ответ и обрабатываем
}
public static void main(String[] args) {
// создаем единую нить запросов
final Queue<CompPart> commonQuery = new LinkedList<CompPart>();
// создаем обработчики по числу процессоров на компьютере
for (int i = 0; i < PROCESSORS; i++) {
Thread th = new Thread(new Processor2(commonQuery)); //создаем нить обработчика
th.setDaemon(true);
th.start();
}
// создаем 5-ть параллельных задач на обработку, разделяющих одну очередь и одних и тех же обработчиков
for (int i = 0; i < 5; i++) {
new Thread(new ParallelComputation2(commonQuery), String.valueOf(i)).start();
}
}
/**
* класс-обработчик
*/
static class Processor2 implements Runnable{
final Queue<CompPart> income; // входная очередь
/**
* конструктор
* @param income
*/
Processor2(Queue<CompPart> income) {
this.income = income;
}
/**
* основной метод ожидания/обработки
*/
public void run() {
try {
// бесконечный цикл можно прервать поставив нить данного обработчика в .interrupt()
// но так как она daemon, можно забить и просто выйти из программы.
while(!Thread.interrupted()) {
CompPart cp = null; // подготавливаем объект для извлечения запроса на обработку
synchronized (income) { // захват очерди
// Если очередь пустая - спим на ней, пока вычислители не накидают данные и не разбудят
if (income.isEmpty()) {
income.wait();
}
cp = income.poll(); // извлекаем объект
}
// может получиться так, что был кинут один объект, а на очереди спало обработчиков
// объект будет захвачен одним обработчиком, остальные получат null и проскочат цикл
if (cp != null) {
double[] matrixRow = cp.getDataRef()[cp.getRow()]; // обрабатываемый ряд
{
}
// process....
for (int i = 0; i < matrixRow.length; i++) {
matrixRow[i] += 2;
}
// inform вычислителя
final ParallelComputation2 pc2;
pc2 = cp.getPc2();
synchronized (pc2) { // захват вычислителя
pc2.countDown--; // уменьшаем переменную обратного отсчета
if (pc2.countDown == 0) { // если больше от вычислителя задач не будет...
pc2.notify(); // будим его (если он спит)
}
}
}
}
} catch (InterruptedException ignored) {}
}
}
/**
* дополнительный класс-сообщение для передачи по очереди.
*/
static class CompPart {
private int row;
private double[][] dataRef;
final private ParallelComputation2 pc2;
CompPart(int row, double[][] dataRef, ParallelComputation2 pc2) {
this.row = row;
this.dataRef = dataRef;
this.pc2 = pc2;
}
public int getRow() {
return row;
}
public double[][] getDataRef() {
return dataRef;
}
public ParallelComputation2 getPc2() {
return pc2;
}
}
}
|
|