Java може да паралелизира поточните операции, за да използва многоядрени системи. Тази статия предоставя перспектива и показва как паралелният поток може да подобри производителността с подходящи примери.
Потоци в Java
Поток в Java е поредица от обекти, представени като проводник от данни. Обикновено има източник където се намират данните и дестинация където се предава. Имайте предвид, че потокът не е хранилище; вместо това той работи с източник на данни, като например масив или колекция. Между битовете в пасажа всъщност се наричат поток. По време на процеса на предаване потокът обикновено преминава през една или повече възможни трансформации, като филтриране или сортиране, или може да бъде всеки друг процес, опериращ с данните. Това персонализира оригиналните данни в различна форма, обикновено според нуждите на програмиста. Следователно се създава нов поток според приложената върху него операция. Например, когато поток се сортира, това води до нов поток, който произвежда резултат, който след това се сортира. Това означава, че новите данни са трансформирано копие на оригинала, а не в оригиналния вид.
Последователен поток
Всяка операция с поток в Java, освен ако не е посочена изрично като паралелна, се обработва последователно. Те са основно непаралелни потоци, използвани с една нишка за обработка на своя тръбопровод. Последователните потоци никога не се възползват от многоядрената система, дори ако основната система може да поддържа паралелно изпълнение. Какво се случва, например, когато приложим многонишковост за обработка на потока? Дори тогава той работи на едно ядро в даден момент. Въпреки това, той може да прескача от едно ядро в друго, освен ако не е изрично закрепено към конкретно ядро. Например, обработката в четири различни нишки спрямо четири различни ядра очевидно е различна, когато първото не съвпада с второто. Напълно възможно е да се изпълняват множество нишки в една среда на ядро, но паралелната обработка е съвсем различен жанр. Една програма трябва да бъде проектирана на базата на паралелно програмиране, освен да се изпълнява в среда, която я поддържа. Това е причината паралелното програмиране да е сложна арена.
Нека опитаме пример, за да илюстрираме идеята допълнително.
package org.mano.example;
import java.util.Arrays;
import java.util.List;
public class Main2 {
public static oid main(String[] args) {
List<Integer> list=Arrays.asList(1,2,3,4,5,6,7,8,9);
list.stream().forEach(System.out::println);
System.out.println();
list.parallelStream().forEach(System.out::println);
}
}
Изход
123456789 685973214
Този пример е илюстрация на q последователен поток, както и на q паралелен поток в действие. list.stream() работи последователно в една нишка с println() операция. list.parallelStream() , от друга страна, се обработва паралелно, като се възползва напълно от основната многоядрена среда. Интересният аспект е в изхода на предходната програма. В случай на последователен поток, съдържанието на списъка се отпечатва в подредена последователност. Изходът на паралелния поток, от друга страна, е неподреден и последователността се променя всеки път, когато програмата се изпълнява. Това означава поне едно нещо:това извикване на list.parallelStream() методът прави println операторът работи в множество нишки, нещо, което list.stream() прави в една нишка.
Паралелен поток
Основната мотивация зад използването на паралелен поток е обработката на поток да стане част от паралелното програмиране, дори ако цялата програма може да не е паралелна. Паралелният поток използва многоядрени процесори, което води до значително увеличение на производителността. За разлика от всяко паралелно програмиране, те са сложни и податливи на грешки. Въпреки това, библиотеката за потоци на Java предоставя възможността да го направите лесно и по надежден начин. Цялата програма може да не е успоредна. но поне частта, която обработва потока, може да бъде паралелизирана. Те всъщност са доста прости в смисъл, че можем да извикаме няколко метода, а за останалото се погрижим. Има няколко начина да го направите. Един такъв начин е да получите паралелен поток чрез извикване на parallelStream() метод, дефиниран от Колекция . Друг начин е да извикате parallel() метод, дефиниран от BaseStream на последователен поток. Последователният поток се паралелизира от извикването. Имайте предвид, че основната платформа трябва да поддържа паралелно програмиране, като например многоядрена система. В противен случай няма смисъл от призоваването. В такъв случай потокът ще бъде обработен последователно, дори ако сме направили извикването. Ако извикването е направено на вече паралелен поток, то не прави нищо и просто връща потока.
За да се гарантира, че резултатът от паралелната обработка, приложена върху потока, е същият като получен чрез последователна обработка, паралелните потоци трябва да са без състояние, да не се намесват и да са асоциативни.
Бърз пример
package org.mano.example;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
public class Main {
public static void main(String[] args) {
List<Employee> employees = Arrays.asList(
new Employee(1276, "FFF",2000.00),
new Employee(7865, "AAA",1200.00),
new Employee(4975, "DDD",3000.00),
new Employee(4499, "CCC",1500.00),
new Employee(9937, "GGG",2800.00),
new Employee(5634, "HHH",1100.00),
new Employee(9276, "BBB",3200.00),
new Employee(6852, "EEE",3400.00));
System.out.println("Original List");
printList(employees);
// Using sequential stream
long start = System.currentTimeMillis();
List<Employee> sortedItems = employees.stream()
.sorted(Comparator
.comparing(Employee::getName))
.collect(Collectors.toList());
long end = System.currentTimeMillis();
System.out.println("sorted using sequential stream");
printList(sortedItems);
System.out.println("Total the time taken process :"
+ (end - start) + " milisec.");
// Using parallel stream
start = System.currentTimeMillis();
List<Employee> anotherSortedItems = employees
.parallelStream().sorted(Comparator
.comparing(Employee::getName))
.collect(Collectors.toList());
end = System.currentTimeMillis();
System.out.println("sorted using parallel stream");
printList(anotherSortedItems);
System.out.println("Total the time taken process :"
+ (end - start) + " milisec.");
double totsal=employees.parallelStream()
.map(e->e.getSalary())
.reduce(0.00,(a1,a2)->a1+a2);
System.out.println("Total Salary expense: "+totsal);
Optional<Employee> maxSal=employees.parallelStream()
.reduce((Employee e1, Employee e2)->
e1.getSalary()<e2.getSalary()?e2:e1);
if(maxSal.isPresent())
System.out.println(maxSal.get().toString());
}
public static void printList(List<Employee> list) {
for (Employee e : list)
System.out.println(e.toString());
}
}
package org.mano.example;
public class Employee {
private int empid;
private String name;
private double salary;
public Employee() {
super();
}
public Employee(int empid, String name,
double salary) {
super();
this.empid = empid;
this.name = name;
this.salary = salary;
}
public int getEmpid() {
return empid;
}
public void setEmpid(int empid) {
this.empid = empid;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public double getSalary() {
return salary;
}
public void setSalary(double salary) {
this.salary = salary;
}
@Override
public String toString() {
return "Employee [empid=" + empid + ", name="
+ name + ", salary=" + salary + "]";
}
}
В предишния код забележете как сме приложили сортиране към поток чрез използване на последователно изпълнение.
List<Employee> sortedItems = employees.stream()
.sorted(Comparator
.comparing(Employee::getName))
.collect(Collectors.toList());
и паралелното изпълнение се постига чрез лека промяна на кода.
List<Employee> anotherSortedItems = employees
.parallelStream().sorted(Comparator
.comparing(Employee::getName))
.collect(Collectors.toList());
Също така ще сравним системното време, за да добием представа коя част от кода отнема повече време. Паралелната операция започва, след като паралелният поток бъде изрично получен от parallelStream() метод. Има и друг интересен метод, наречен reduce() . Когато приложим този метод към паралелен поток, операцията може да се извърши в различни нишки.
Въпреки това винаги можем да превключваме между паралелно и последователно според нуждите. Ако искаме да променим паралелния поток на последователен, можем да го направим, като извикаме sequential() метод, посочен от BaseStream . Както видяхме в първата ни програма, операцията, извършена върху потока, може да бъде подредена или неподредена според реда на елементите. Това означава, че поръчката зависи от източника на данни. Това обаче не е положението в случая на паралелни потоци. За да се повиши производителността, те се обработват паралелно. Тъй като това се прави без никаква последователност, където всеки дял от потока се обработва независимо от другите дялове без никаква координация, последствието е непредвидимо неподредено. Но ако искаме конкретно да извършим операция върху всеки елемент в паралелния поток, който да бъде подреден, можем да разгледаме forEachOrdered() метод, който е алтернатива на forEach() метод.
Заключение
Поточните API са част от Java от дълго време, но добавянето на настройката на паралелната обработка е много приветливо и в същото време доста интригуваща функция. Това е особено вярно, защото съвременните машини са многоядрени и има клеймо, че дизайнът на паралелно програмиране е сложен. API, предоставени от Java, предоставят възможността за включване на нотка на настройки за паралелно програмиране в програма на Java, която има цялостния дизайн на последователно изпълнение. Това е може би най-добрата част от тази функция.
