Java puede paralelizar las operaciones de transmisión para aprovechar los sistemas de varios núcleos. Este artículo proporciona una perspectiva y muestra cómo la transmisión en paralelo puede mejorar el rendimiento con ejemplos apropiados.
Transmisiones en Java
Un arroyo en Java es una secuencia de objetos representados como un conducto de datos. Suele tener una fuente dónde se encuentran los datos y un destino donde se transmite. Tenga en cuenta que una secuencia no es un repositorio; en su lugar, opera en una fuente de datos, como una matriz o una colección. Los bits intermedios en el pasaje en realidad se denominan flujo. Durante el proceso de transmisión, el flujo generalmente pasa por una o más transformaciones posibles, como filtrado o clasificación, o puede ser cualquier otro proceso que opere sobre los datos. Esto personaliza los datos originales en una forma diferente, típicamente, según la necesidad del programador. Por lo tanto, se crea una nueva secuencia de acuerdo con la operación aplicada en ella. Por ejemplo, cuando se ordena un flujo, da como resultado un flujo nuevo que produce un resultado que luego se ordena. Esto significa que los nuevos datos son una copia transformada del original en lugar de estar en la forma original.
Flujo secuencial
Cualquier operación de flujo en Java, a menos que se especifique explícitamente como paralelo, se procesa secuencialmente. Básicamente, son flujos no paralelos que utilizan un solo subproceso para procesar su canalización. Los flujos secuenciales nunca aprovechan el sistema multinúcleo, incluso si el sistema subyacente puede admitir la ejecución en paralelo. ¿Qué sucede, por ejemplo, cuando aplicamos subprocesos múltiples para procesar la transmisión? Incluso entonces, opera en un solo núcleo a la vez. Sin embargo, puede saltar de un núcleo a otro a menos que esté explícitamente anclado a un núcleo específico. Por ejemplo, el procesamiento en cuatro subprocesos diferentes versus cuatro núcleos diferentes es obviamente diferente donde el primero no coincide con el segundo. Es bastante posible ejecutar varios subprocesos en un entorno de un solo núcleo, pero el procesamiento en paralelo es un género completamente diferente. Un programa debe diseñarse desde cero para la programación paralela, además de ejecutarse en un entorno que lo admita. Esta es la razón por la que la programación paralela es un campo complejo.
Probemos un ejemplo para ilustrar más la idea.
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);
}
}
Salida
123456789 685973214
Este ejemplo es una ilustración de q flujo secuencial y q flujo paralelo en funcionamiento. La lista.stream() funciona en secuencia en un solo hilo con println() operación. lista.parallelStream() , por otro lado, se procesa en paralelo, aprovechando al máximo el entorno multinúcleo subyacente. El aspecto interesante está en la salida del programa anterior. En el caso de un flujo secuencial, el contenido de la lista se imprime en una secuencia ordenada. La salida del flujo paralelo, por otro lado, no está ordenada y la secuencia cambia cada vez que se ejecuta el programa. Esto significa al menos una cosa:la invocación de list.parallelStream() hace que el println opera en múltiples subprocesos, algo que list.stream() hace en un solo hilo.
Transmisión paralela
La principal motivación detrás del uso de un flujo paralelo es hacer que el procesamiento de flujo sea parte de la programación paralela, incluso si no se puede paralelizar todo el programa. La transmisión en paralelo aprovecha los procesadores multinúcleo, lo que resulta en un aumento sustancial del rendimiento. A diferencia de cualquier programación paralela, son complejos y propensos a errores. Sin embargo, la biblioteca de flujo de Java proporciona la capacidad de hacerlo fácilmente y de manera confiable. Es posible que no se pueda paralelizar todo el programa. pero al menos la parte que maneja la transmisión se puede paralelizar. En realidad, son bastante simples en el sentido de que podemos invocar algunos métodos y el resto está resuelto. Hay un par de maneras de hacerlo. Una de estas formas es obtener un flujo paralelo invocando parallelStream() método definido por Colección . Otra forma es invocar el parallel() método definido por BaseStream en un flujo secuencial. El flujo secuencial es paralelizado por la invocación. Tenga en cuenta que la plataforma subyacente debe admitir la programación paralela, como con un sistema multinúcleo. De lo contrario, no tiene sentido la invocación. La transmisión se procesaría en secuencia en tal caso, incluso si hemos realizado la invocación. Si la invocación se realiza en un flujo ya paralelo, no hace nada y simplemente devuelve el flujo.
Para garantizar que el resultado del procesamiento paralelo aplicado en la transmisión sea el mismo que se obtiene a través del procesamiento secuencial, las transmisiones paralelas deben ser sin estado, sin interferencias y asociativas.
Un ejemplo rápido
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 + "]";
}
}
En el código anterior, observe cómo hemos aplicado la ordenación en una secuencia usando la ejecución secuencial.
List<Employee> sortedItems = employees.stream()
.sorted(Comparator
.comparing(Employee::getName))
.collect(Collectors.toList());
y la ejecución paralela se logra cambiando ligeramente el código.
List<Employee> anotherSortedItems = employees
.parallelStream().sorted(Comparator
.comparing(Employee::getName))
.collect(Collectors.toList());
También compararemos el tiempo del sistema para tener una idea de qué parte del código lleva más tiempo. La operación paralela comienza una vez que parallelStream() obtiene explícitamente el flujo paralelo. método. Hay otro método interesante, llamado reduce() . Cuando aplicamos este método a un flujo paralelo, la operación puede ocurrir en diferentes subprocesos.
Sin embargo, siempre podemos cambiar entre paralelo y secuencial según la necesidad. Si queremos cambiar el flujo paralelo a secuencial, podemos hacerlo invocando el secuencial() método especificado por BaseStream . Como vimos en nuestro primer programa, la operación realizada sobre el flujo puede ordenarse o desordenarse según el orden de los elementos. Esto significa que el orden depende de la fuente de datos. Esta, sin embargo, no es la situación en el caso de flujos paralelos. Para aumentar el rendimiento, se procesan en paralelo. Debido a que esto se hace sin ninguna secuencia, donde cada partición del flujo se procesa independientemente de las otras particiones sin ninguna coordinación, la consecuencia es un desorden impredecible. Pero, si queremos realizar específicamente una operación en cada elemento del flujo paralelo que se va a ordenar, podemos considerar el forEachOrdered() método, que es una alternativa al método forEach() método.
Conclusión
Las API de transmisión han sido parte de Java durante mucho tiempo, pero agregar la modificación del procesamiento paralelo es muy agradable y, al mismo tiempo, es una característica bastante intrigante. Esto es particularmente cierto porque las máquinas modernas son multinúcleo y existe el estigma de que el diseño de programación paralela es complejo. Las API proporcionadas por Java brindan la capacidad de incorporar un matiz de ajustes de programación paralela en un programa Java que tiene el diseño general de ejecución secuencial. Esta es quizás la mejor parte de esta característica.
