От  Fork/Join  к Stream API

Таблицы с данными

Таблицы с данными
2. Сущность «договор», по которой необходимо собрать, обработать

Таблицы с данными
2. Сущность «договор», по которой необходимо собрать, обработать

на поверхности
1. На основе исходных данных создать «подготовленные» промежуточные таблицы

на поверхности
1. На основе исходных данных создать «подготовленные» промежуточные таблицы

на поверхности
1. На основе исходных данных создать «подготовленные» промежуточные таблицы

ли концепция Fork/Join для вычислений?

Fork/Join для вычислений?
Насколько легко использовать данные конструкции?

концепция Fork/Join для вычислений?
Насколько легко использовать данные конструкции?
Ошибки будут?

концепция Fork/Join для вычислений?
Насколько легко использовать данные конструкции?
Ошибки будут?
Кто

С его помощью можно довольно просто использовать возможности процессоров, доступных

С его помощью можно довольно просто использовать возможности процессоров, доступных

С его помощью можно довольно просто использовать возможности процессоров, доступных

понятием «threshold»
В переводе – это порог, предел, пороговый уровень

В литературе

RecursiveAction {

@Override
protected void compute() { if (не достигли

static void main(String[] args) {
// расчеты

«Делим» (2)

import java.util.concurrent.RecursiveAction; public class Example extends RecursiveAction { int cuntPr = Runtime.getRuntime().availableProcessors();
// условный threshold (предел) int cntLimit = 3 000 000; int start; int end; Example(int startNumber, int endNumber) { start = startNumber; end = endNumber; } protected void compute() { if (end - start <= cntLimit) { for (int i = start; i <= end; i++) { new Calc().go(i); } } else { System.out.println("=split="); int middle = (start + end) / 2; invokeAll(new Example(start, middle), new Example(middle + 1, end)); } } }

Вычисления. Возводим число в заданную степень (3)

public class Calc { public void go(int numberForCalc) { for(int i = 0; i <= numberForCalc; i++) { double pow = Math.pow(numberForCalc,45); } } }

args) {
// расчеты до 7 000

cuntPr = Runtime.getRuntime().availableProcessors();
//

= ……………..
NO ERROR
=split=
Time = ……………..
NO ERROR



В)
=split=
=split=
=split=
Time = ……………..
NO ERROR


Г)

{ int componentValue = 7 000 000; ForkJoinPool fjp = new ForkJoinPool();
Example

Создание пула задач
ForkJoinPool fjp = new ForkJoinPool();
Создание пула задач начиная с JDK 8
ForkJoinPool fjp = ForkJoinPool.commonPool();
Уровень параллелизма
а) поставить по умолчанию (будет зависеть от
количества процессоров);
Runtime.getRuntime().availableProcessors();
ВАЖНО: фактически это - number of logical cores
б) указать кол-во в ForkJoinPool
(int уровень_параллелизма);

Из документации
ForkJoinPool Creates a ForkJoinPool with parallelism
equal to Runtime.availableProcessors(), using the default
thread factory, no UncaughtExceptionHandler, and
non-async LIFO processing mode

ForkJoinPool (int parallelism) Creates
a ForkJoinPool with the indicated parallelism level,
the default thread factory, no UncaughtExceptionHandler,
and non-async LIFO processing mode

задачи
Например, умножим N (количество элементов) на Q (стоимость на

задачи
Например, умножим N (количество элементов) на Q (стоимость на

000
threshold = 3 000 000

Вариант №2 (threshold = 7 000

on a Sun Fire T2000 server from Oracle where the
number

массиве из 500 000 элементов
на различных системах и с различными

зависит от БД

зависит от БД

3. В большинстве случаев не требуется
закупка дополнительного оборудования

зависит от БД

3. В большинстве случаев не требуется
закупка дополнительного оборудования

4.

зависит от БД

3. В большинстве случаев не требуется
закупка дополнительного оборудования

4.

зависит от БД

3. В большинстве случаев не требуется
закупка дополнительного оборудования

4.

порогового значения (threshold)
Решается перебором возможных значений с учетом
возможностей аппаратных средств.

порогового значения (threshold)
2. Проблема контроля загруженности ядер процессора и выбор

порогового значения (threshold)
2. Проблема контроля загруженности ядер процессора и выбор

порогового значения (threshold)
2. Проблема контроля загруженности ядер процессора и выбор

порогового значения (threshold)
2. Проблема контроля загруженности ядер процессора и выбор

порогового значения (threshold)
2. Проблема контроля загруженности ядер процессора и выбор

полученные данные

Получить часть данных до выполнения расчета

Улучшить читаемость кода

Обработку части

полученные данные

Получить часть данных до выполнения расчета

Улучшить читаемость кода

Обработку

Из документации
«Another implementation of the
fork/join framework is used by methods
in the java.util.streams package, which
is part of Project Lambda scheduled
for the Java SE 8 release. For more
information, see the Lambda
Expressions section.»

Collection

Для использования parallelStream() нужно понимать как он работает

Перед

and other sources.
Most stream methods apply the
same operation to

Ответ
Consider using S.parallelStream().operation(F) instead of S.stream().operation(F) when operations are independent, and either computationally expensive or applied to many elements of efficiently splittable data structures, or both. In more detail:F, the per-element function (usually a lambda) is independent: the computation for each element does not rely on or impact that of any other element. (See the stream package summary for further guidance about using stateless non-interfering functions.)S, the source collection is efficiently splittable. There are a few other readily parallelizable stream sources besides Collections, for example, java.util.SplittableRandom (for which you can use the stream.parallel() method to parallelize). But most sources based on IO are designed primarily for sequential use.The total time to execute the sequential version exceeds a minimum threshold. These days, the threshold is roughly (within a factor of ten of) 100 microseconds across most platforms. You don't need to measure this precisely though. You can estimate this well enough in practice by multiplying N (the number of elements) by Q (cost per element of F), in turn guestimating Q as the number of operations or lines of code, and then checking that N * Q is at least 10000. (If you are feeling cowardly, add another zero or two.) So when F is a tiny function like x -> x + 1, then it would require N >= 10000 elements for parallel execution to be worthwhile. And conversely, when F is a massive computation like finding the best next move in a chess game, the Q factor is so high that N doesn't matter so long as the collection is completely splittable. The streams framework does not (and cannot) enforce any of these. If the computation is not independent, then running it in parallel will not make any sense and might even be harmfully wrong.

понять, когда можно
безболезненно использовать parallelStream().
Ответ Doug Lea
Умножим N (количество элементов)

в своей книге
говорит о теме параллельности и
пороговых значениях.

Данная

«При замере времени работы примеров
6.1 и 6.2 на 4-ядерной машине при 10
Альбомах последовательная версия
оказывается в 8 раз быстрее. При 100
альбомах обе версии работают одинаково
быстро, а при 10 000 альбомов 
параллельная версия опережает
последовательную в 2,5 раза. Все результаты измерений в этой главе
приводятся только для сведения. На
вашей машине они могут оказаться
совершенно другими. Размер входного
потока – не единственный фактор,
определяющий, даст ли распараллеливание
ускорение. Результаты могут также зависеть
от способа написания кода и количества
доступных ядер.»

public int serialArraySum() {
return albums.stream()
.flatMap(Album::getTracks)
.mapToInt(Track::getLength)
.sum();
}








Пример «6.2»


public int parallelArraySum() {
return

Валеев. В сети можно найти его
статьи и видео по

«Ваша работа — это N*Q, где N —
количество элементов, а Q - среднее
время обработки элемента. Если Q
очень велико, то распараллеливать
имеет смысл уже при N =2. Если Q
исключительно мало (например,
сложение двух чисел), то
параллелизм поможет только при
больших N. Кроме того, многое
зависит от самих операций
в стриме. Если все операции
stateless, вы можете получить
хороший прирост.
Если имеются stateful-операции,
можно сильно замедлиться даже
для очень большого N. Переменных
очень много.»