Java dapat memparalelkan operasi aliran untuk memanfaatkan sistem multi-inti. Artikel ini memberikan perspektif dan menunjukkan bagaimana streaming paralel dapat meningkatkan performa dengan contoh yang sesuai.
Streaming di Java
Sebuah aliran di Jawa adalah urutan objek yang direpresentasikan sebagai saluran data. Biasanya memiliki sumber tempat data berada dan tujuan di mana itu ditransmisikan. Perhatikan bahwa aliran bukanlah repositori; sebagai gantinya, ini beroperasi pada sumber data seperti pada larik atau kumpulan. Bit-bit di antara dalam bagian itu sebenarnya disebut aliran. Selama proses transmisi, aliran biasanya melewati satu atau lebih transformasi yang mungkin, seperti penyaringan atau pengurutan, atau bisa juga proses lain yang beroperasi pada data. Ini menyesuaikan data asli ke dalam bentuk yang berbeda, biasanya, sesuai dengan kebutuhan programmer. Oleh karena itu, aliran baru dibuat sesuai dengan operasi yang diterapkan padanya. Misalnya, ketika aliran diurutkan, itu menghasilkan aliran baru yang menghasilkan hasil yang kemudian diurutkan. Ini berarti data baru adalah salinan yang diubah dari aslinya, bukan dalam bentuk aslinya.
Aliran Berurutan
Setiap operasi aliran di Java, kecuali secara eksplisit ditentukan sebagai paralel, diproses secara berurutan. Mereka pada dasarnya adalah aliran non-paralel yang menggunakan satu utas untuk memproses pipa mereka. Aliran sekuensial tidak pernah mengambil keuntungan dari sistem multicore bahkan jika sistem yang mendasarinya dapat mendukung eksekusi paralel. Apa yang terjadi, misalnya, ketika kita menerapkan multithreading untuk memproses aliran? Meski begitu, ia beroperasi pada satu inti pada satu waktu. Namun, itu mungkin melompat dari satu inti ke inti lainnya kecuali secara eksplisit disematkan ke inti tertentu. Misalnya, pemrosesan dalam empat utas berbeda versus empat inti berbeda jelas berbeda di mana yang pertama tidak cocok dengan yang terakhir. Sangat mungkin untuk mengeksekusi beberapa utas dalam lingkungan inti tunggal tetapi pemrosesan paralel adalah genre yang berbeda sama sekali. Sebuah program perlu dirancang untuk pemrograman paralel selain dari mengeksekusi di lingkungan yang mendukungnya. Inilah alasan mengapa pemrograman paralel adalah arena yang kompleks.
Mari kita coba contoh untuk mengilustrasikan ide lebih lanjut.
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);
}
}
Keluaran
123456789 685973214
Contoh ini adalah ilustrasi dari q aliran sekuensial serta q aliran paralel yang beroperasi. list.stream() bekerja secara berurutan pada satu utas dengan println() operasi. list.parallelStream() , di sisi lain, diproses secara paralel, mengambil keuntungan penuh dari lingkungan multicore yang mendasarinya. Hal yang menarik adalah pada output dari program sebelumnya. Dalam kasus aliran berurutan, konten daftar dicetak dalam urutan yang berurutan. Output dari aliran paralel, di sisi lain, tidak berurutan dan urutannya berubah setiap kali program dijalankan. Ini menandakan setidaknya satu hal:pemanggilan list.parallelStream() metode membuat println pernyataan beroperasi di banyak utas, sesuatu yang list.stream() dilakukan dalam satu utas.
Aliran Paralel
Motivasi utama di balik penggunaan aliran paralel adalah menjadikan pemrosesan aliran sebagai bagian dari pemrograman paralel, bahkan jika keseluruhan program mungkin tidak diparalelkan. Aliran paralel memanfaatkan prosesor multicore, menghasilkan peningkatan kinerja yang substansial. Tidak seperti pemrograman paralel lainnya, program ini kompleks dan rawan kesalahan. Namun, perpustakaan aliran Java menyediakan kemampuan untuk melakukannya dengan mudah, dan dengan cara yang andal. Seluruh program mungkin tidak diparalelkan. tapi setidaknya bagian yang menangani aliran bisa diparalelkan. Mereka sebenarnya cukup sederhana dalam arti bahwa kita dapat memanggil beberapa metode dan sisanya diurus. Ada beberapa cara untuk melakukannya. Salah satu caranya adalah mendapatkan aliran paralel dengan menjalankan parallelStream() metode yang ditentukan oleh Koleksi . Cara lain adalah dengan memanggil parallel() metode yang ditentukan oleh BaseStream pada aliran berurutan. Aliran sekuensial diparalelkan dengan pemanggilan. Perhatikan bahwa platform yang mendasari harus mendukung pemrograman paralel, seperti dengan sistem multicore. Kalau tidak, tidak ada gunanya berdoa. Aliran akan diproses secara berurutan dalam kasus seperti itu, bahkan jika kita telah membuat permintaan. Jika pemanggilan dibuat pada aliran yang sudah paralel, itu tidak melakukan apa-apa dan hanya mengembalikan aliran.
Untuk memastikan bahwa hasil pemrosesan paralel yang diterapkan pada aliran sama dengan yang diperoleh melalui pemrosesan berurutan, aliran paralel harus stateless, tidak mengganggu, dan asosiatif.
Contoh Singkat
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 + "]";
}
}
Pada kode sebelumnya, perhatikan bagaimana kami telah menerapkan pengurutan pada aliran dengan menggunakan eksekusi berurutan.
List<Employee> sortedItems = employees.stream()
.sorted(Comparator
.comparing(Employee::getName))
.collect(Collectors.toList());
dan eksekusi paralel dicapai dengan sedikit mengubah kode.
List<Employee> anotherSortedItems = employees
.parallelStream().sorted(Comparator
.comparing(Employee::getName))
.collect(Collectors.toList());
Kami juga akan membandingkan waktu sistem untuk mendapatkan ide bagian kode mana yang membutuhkan lebih banyak waktu. Operasi paralel dimulai setelah aliran paralel diperoleh secara eksplisit oleh parallelStream() metode. Ada metode lain yang menarik, yang disebut reduce() . Saat kami menerapkan metode ini ke aliran paralel, operasi dapat terjadi di utas yang berbeda.
Namun, kami selalu dapat beralih antara paralel dan berurutan sesuai kebutuhan. Jika kita ingin mengubah aliran paralel menjadi sekuensial, kita dapat melakukannya dengan memanggil sequential() metode yang ditentukan oleh BaseStream . Seperti yang kita lihat di program pertama kami, operasi yang dilakukan pada aliran dapat dipesan atau tidak diurutkan sesuai dengan urutan elemen. Ini berarti bahwa urutannya tergantung pada sumber data. Namun, ini bukan situasi dalam kasus aliran paralel. Untuk meningkatkan kinerja, mereka diproses secara paralel. Karena ini dilakukan tanpa urutan apa pun, di mana setiap partisi aliran diproses secara independen dari partisi lain tanpa koordinasi apa pun, konsekuensinya tidak dapat diprediksi secara tidak berurutan. Namun, jika kita ingin secara khusus melakukan operasi pada setiap elemen dalam aliran paralel yang akan dipesan, kita dapat mempertimbangkan forEachOrdered() metode, yang merupakan alternatif dari forEach() metode.
Kesimpulan
API aliran telah menjadi bagian dari Java untuk waktu yang lama, tetapi menambahkan tweak pemrosesan paralel sangat menyambut, dan pada saat yang sama merupakan fitur yang cukup menarik. Hal ini terutama benar karena mesin modern multicore dan ada stigma bahwa desain pemrograman paralel itu rumit. API yang disediakan oleh Java menyediakan kemampuan untuk menggabungkan sedikit tweak pemrograman paralel dalam program Java yang memiliki desain keseluruhan dari eksekusi sekuensial. Ini mungkin bagian terbaik dari fitur ini.
