Стратегии параллельных конвейеров в Java

Учитывая, что вы не говорите, как вы измеряете прошедшее время, я предполагаю, что вы используете что-то вроде:

time java org.egge.concurrent.SerialTest < in.gz > out.gz
time java org.egge.concurrent.ConcurrentQueueTest < in.gz > out.gz

проблема с этим, что вы измеряете две вещи здесь:

1. Как долго JVM принимает, чтобы начать, а
2. Как долго требуется программа для запуска.

Вы можете изменить только второй с изменениями кода. Используя цифры, которые вы дали:

Testing SerialTest
real    0m6.736s
user    0m6.924s
sys     0m0.245s

Testing ParallelExchangerTest
real    0m4.967s
user    0m7.491s
sys     0m0.850s

, если мы предполагаем, что запуск JVM занимает три секунды, то «время выполнения программы» составляет 3,7 и 1,9 секунды соответственно, это в значительной степени ускорение 100%. Я решительно предположим, что вы используете большой набор набора данных для тестирования, чтобы вы могли минимизировать влияние запуска JVM в результате вашего времени.

Редактировать : На основании ваших ответов на этот вопрос вы вполне можете страдать от конфликта блокировки. Лучший способ решить, что в Java, вероятно, использовать трубопроводные читатели и писатели, прочитанные с труб, байт за раз, и заменить любые @ ' символов в входном потоке с помощью «_AT _» в выходном потоке. Вы можете страдать от того факта, что каждая строка отсканирована три раза, и любая замена требует, чтобы новый объект был создан, и строка заканчивается снова скопированным. Надеюсь, это поможет ...

На Java тоже можно использовать трубы. Они реализованы в виде потоков, подробнее см. PipedInputStream и PipedOutputStream.

Чтобы предотвратить блокировку, я бы порекомендовал установить размер пропперной трубы.

Уменьшение количества читателей и объектов дают мне более чем на 10% лучшую производительность.

Но производительность Java.util.concurrent еще немного разочаровывает.

ConcularentQueetest:

private static class Reader implements Runnable {

@Override
  public void run() {
   final char buf[] = new char[8192];
   try {

    int len;
    while ((len = reader.read(buf)) != -1) {
     pipe.put(new String(buf,0,len));
    }
    pipe.put(POISON);

   } catch (IOException e) {
    throw new RuntimeException(e);
   } catch (InterruptedException e) {
    throw new RuntimeException(e);
   }
  }

Во-первых, процесс будет таким же быстрым, как и самый медленный фрагмент. Если разбивка по времени выглядит так:

• gunzip: 1 секунда
• sed: 5 секунд
• gzip: 1 секунда

, при многопоточности вы будете работать в в лучшем случае 5 секунд вместо 7.

Во-вторых, вместо использования очередей, которые вы используете, вместо этого попытайтесь воспроизвести функциональность того, что вы копируете, и используйте PipedInputStream и PipedOutputStream для объединить процессы.

Изменить: Есть несколько способов обработки связанных задач с помощью утилит параллелизма Java. Разделите его на нитки. Сначала создайте общий базовый класс:

public interface Worker {
  public run(InputStream in, OutputStream out);
}

Этот интерфейс представляет некоторую произвольную работу, которая обрабатывает ввод и генерирует вывод. Соедините их вместе, и у вас будет конвейер. Вы также можете абстрагироваться от шаблона. Для этого нам понадобится класс:

public class UnitOfWork implements Runnable {
  private final InputStream in;
  private final OutputStream out;
  private final Worker worker;

  public UnitOfWork(InputStream in, OutputStream out, Worker worker) {
    if (in == null) {
      throw new NullPointerException("in is null");
    }
    if (out == null) {
      throw new NullPointerException("out is null");
    }
    if (worker == null) {
      throw new NullPointerException("worker is null");
    }
    this.in = in;
    this.out = out;
    this.worker = worker;
  }

  public final void run() {
    worker.run(in, out);
  }
}

Так, например, Unzip PART:

public class Unzip implements Worker {
  protected void run(InputStream in, OutputStream out) {
    ...
  }
}

и так далее для Sed и Zip . Что затем связывает это воедино, так это следующее:

public static void pipe(InputStream in, OutputStream out, Worker... workers) {
  if (workers.length == 0) {
    throw new IllegalArgumentException("no workers");
  }
  OutputStream last = null;
  List<UnitOfWork> work = new ArrayList<UnitOfWork>(workers.length);
  PipedOutputStream last = null;
  for (int i=0; i<workers.length-2; i++) {
    PipedOutputStream out = new PipedOutputStream();
    work.add(new UnitOfWork(
      last == null ? in, new PipedInputStream(last), out, workers[i]);
    last = out;
  }
  work.add(new UnitOfWork(new PipedInputStream(last),
    out, workers[workers.length-1);
  ExecutorService exec = Executors.newFixedThreadPool(work.size());
  for (UnitOfWork w : work) {
    exec.submit(w);
  }
  exec.shutdown();
  try {
    exec.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
  } catch (InterruptedExxception e) {
    // do whatever
  }
}

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

public static processFile(String inputName, String outputName) {
  pipe(new FileInputStream(inputFile), new FileOutputStream(outputFile),
    new Zip(), new Sed(), new Unzip());
}