¿Cómo implementa RxJava múltiples subprocesos que se ejecutan al mismo tiempo y cómo implementar el reintento?

Al escribir una clase, si el código de la clase puede ejecutarse en un entorno de subprocesos múltiples, se deben considerar los problemas de sincronización. Hay muchas formas de lograr la sincronización de subprocesos en Java, como se muestra a continuación.

(1) palabra clave sincronizada

La palabra clave sincronizada primitiva de sincronización a nivel de lenguaje está integrada en Java, que realiza el control de recursos compartidos en condiciones de acceso sincrónico. Dependiendo de los objetos modificados por la palabra clave sincronizada, se puede dividir en las siguientes situaciones.

*método de sincronización de palabras clave sincronizadas

método público sincronizado void(){

//hacer algo

}

Nota: Si utiliza el método de sincronización de palabras clave sincronizadas, es fácil pensar erróneamente que la palabra clave sincronizada bloquea el código que la rodea. Pero la situación real no es así. Es el objeto el que está bloqueado sincrónicamente, no el código. por lo tanto. Si hay un método sincronizado en una clase, el método puede ser ejecutado por dos subprocesos diferentes al mismo tiempo, siempre que cada subproceso cree una instancia de la clase.

Código de muestra:

paquete newthread;

clase pública TestSync {

public static void main(String[] args) {

MiHilo1 mi1=nuevo MiHilo1(1);

MiHilo1 mi2=nuevo MiHilo1(3);

mi1.start();

my2.start();

}

}

clase MyThread1 extiende Thread{

private int val;

público MyThread1(int v){

val=v;

}

público sincronizado void printVal(int v){

for(int i=0;ilt;100;i){

System.out.print(v);

}

} p>

public void run(){

printVal(val);

}

}

El resultado Las formas de ejecutar el código son salida cruzada 1 y 3, es decir, dos subprocesos 1 y 3 están ejecutando el método printVal al mismo tiempo y la función de sincronización no está implementada. La razón es que la palabra clave sincronizada bloquea el objeto en lugar del bloque de código. Si necesita lograr una verdadera sincronización, debe sincronizar un objeto global o sincronizar la clase.

El formato de la clase de sincronización de palabras clave sincronizadas es el siguiente:

synchronized(MyThread.class){}

Mejorar el código

paquete newthread;

clase pública TestSync_2 {

public static void main(String[] args) {

MyThread_2 my1=new MyThread_2(1);

my1 .start() ;

MiHilo_2 my2=nuevo MiHilo_2(2);

my2.start();

}

}

clase MyThread_2 extiende Thread{

private int val;

public MyThread_2(int v){

val=v;

}

public void printVal(int v){

sincronizado(MyThread_2.class){

for(int i=0; ilt ; 100; yo ){

Sistema.out.print(v);

}

}

}

public void run(){

printVal(val);

}

}

En el ejemplo anterior, el Método printVal() La función ya no sincroniza clases individuales, sino que sincroniza la clase actual. Para MyThread, solo tiene una definición de clase única. Los dos subprocesos están sincronizados en el mismo bloqueo, por lo que solo un subproceso puede ejecutar el método printVal () al mismo tiempo. En cuanto a las dos posibilidades de resultados de salida, están determinadas por el modo de implementación preferente de la programación de subprocesos de Java.

La palabra clave * sincronizada sincroniza las variables miembro estáticas del público ***

En el ejemplo anterior, al bloquear la clase actual, se logra el efecto de sincronización de subprocesos, pero en función on El principio general de la sincronización de subprocesos es que la granularidad de la sincronización debe minimizarse para lograr un mayor rendimiento. De hecho, para el ejemplo anterior, también se puede lograr bloqueando el objeto público, es decir, agregando una variable miembro estática. Ambos métodos logran la seguridad de los subprocesos al sincronizar el objeto.

El código de muestra es el siguiente:

paquete newthread;

clase pública TestSync_3 {

public static void main(String[] args) {

MiHilo_3 mi1=nuevo MiHilo_3(2);

my1.start();

MiHilo_3 mi2=nuevo MiHilo_3(5);

mi2. start() ;

}

}

clase MyThread_3 extiende Thread{

private int val;

bloqueo de objeto estático privado = nuevo objeto();

public MyThread_3(int v){

val=v; public void printVal (int v){

sincronizado(lock){

for(int i=0;ilt;100;i){

System. out.print (v);

}

}

}

public void run(){

printVal( val);

}

}

Nota: Para mejorar el rendimiento del programa, se realizará una breve introducción a la selección de objetos en se proporciona el código de muestra: Basado en el mecanismo de optimización de la JVM Dado que los objetos de tipo Cadena son inmutables, cuando el usuario usa la forma "" para referirse a una cadena, si la JVM encuentra que ya existe dicho objeto en la memoria. , utilizará el objeto en lugar de generar uno. Se utiliza un nuevo objeto String para reducir el uso de memoria; crear un objeto de matriz de bytes de longitud cero es más práctico que cualquier otro objeto. Solo requiere 3 códigos de operación para generar un byte de longitud cero. [] objeto y Object lock=new Object() requiere 7 líneas de código de operación.

*la palabra clave sincronizada sincroniza bloques libres

sincronizada{

//hacer algo

}

sincronizada El El bloque de código de sincronización de palabras clave es similar a las variables miembro estáticas del público de sincronización de palabras clave sincronizadas mencionado anteriormente. Todos están diseñados para reducir la granularidad de la sincronización y evitar la sincronización de toda la clase, lo que reduce en gran medida el rendimiento del programa.

*palabra clave sincronizada método estático sincronizado

método vacío estático sincronizado públicoAAA(){

//hacer algo

}

vacío público métodoBBB(){

sincronizado(Test.class){

//hacer algo

}

}

El método estático sincronizado de la palabra clave sincronizada tiene el mismo efecto que la implementación de la clase sincronizada de la palabra clave sincronizada. La única diferencia es que los objetos de bloqueo obtenidos son diferentes. Cuando el mismo objeto accede al método AAA () y al método BBB (), el método estático sincronizado. el bloqueo obtenido es la clase de objeto, y el bloqueo de objeto obtenido por el método de clase de sincronización es la clase a la que pertenece el objeto

(2) Diseño y uso de Metux mutex

Metux Conocido como mutex, también se usa ampliamente en programación multiproceso. . Entre ellos, Mutex implementado en el conjunto de herramientas concurrentes escrito por el profesor Doug Lea es el más típico. Este artículo utilizará esto como ejemplo para presentar brevemente el diseño y el uso de mutex en la programación de subprocesos múltiples. En el conjunto de herramientas concurrentes de Doug Lea, Mutex implementa la interfaz Sync, que es una interfaz pública para todas las cerraduras, puertas y variables de condición en el conjunto de herramientas concurrentes. La clase de implementación de SyncD incluye principalmente Metux, Semaphore y sus subclases, Latch, CountDown y ReentrantLock. , etc. Esto también refleja la idea de programación abstracta orientada a objetos, que permite a los desarrolladores elegir usar diferentes implementaciones de Sync sin cambiar el código o cambiar una pequeña cantidad de código.

La definición de la interfaz de sincronización es la siguiente:

paquete newthread;

interfaz pública Sync {

//Obtener permiso

public void adquirir() lanza InterruptedException;

//Intenta obtener permiso

intento booleano público (mseg largos) lanza InterruptedException;

//Libera permiso

public void realse();

}

La palabra clave sincronizada se puede reemplazar usando la interfaz Sync y proporciona un control de sincronización más flexible, pero esto no significa que el conjunto de herramientas concurrentes es independiente de la tecnología sincronizada; de hecho, el conjunto de herramientas concurrentes también se basa en sincronizado. La diferencia es que la palabra clave sincronizada solo es válida dentro de un método o bloque de código, pero el uso de Sync puede sincronizar entre métodos e incluso entre objetos al pasar entre objetos. Aquí es donde Sync y el conjunto de herramientas concurrentes son más poderosos que usar sincronizado directamente.

Cabe señalar que los métodos adquirir () e intento () en Sync generarán InterruptedException, por lo que cuando use Sync y sus subclases, debe detectar InterruptedException al llamar a estos métodos. El método release() no arroja una excepción interrumpida. Esto se debe a que se puede llamar al método wait() en los métodos adquirir() e try() para esperar a que otros subprocesos liberen el bloqueo. Por lo tanto, no hay ningún problema si se llama al método release() en un hilo que no tiene un método adquirir(). Dado que el método release() no arroja InterruptedException, se debe llamar al método release() en la cláusula catch o finalmente para garantizar que el bloqueo adquirido se pueda liberar correctamente. El código de muestra es el siguiente:

paquete newthread;

clase pública TestSync_4 {

Puerta de sincronización;

prueba pública vacía() {

prueba {

gate.acquire();

prueba{

//haz algo

}finalmente{

gate.realse();

}

} captura (InterruptedException ex) {

}

}

}

Mutex es un bloqueo mutex no reentrante. Se usa ampliamente en entornos de sincronización de tipos anteriores o posteriores que necesitan abarcar métodos.

La siguiente es una implementación de Mutex en el kit de herramientas concurrentes de Doug Lea. El código es el siguiente:

paquete newthread;

importar com.sun.corba.se.impl.orbutil.concurrent. Sync;

la clase pública TestMutex implementa Sync{

protected boolean flg=false;

public void adquirir() lanza InterruptedException {

if(Thread.interrupted())

lanza una nueva InterruptedException;

sincronizado(esto){

prueba{

while( flg )

esperar();

flg=true

}catch(InterruptedException ex){

notificar();

p>

throw ex;

}

}

}

intento booleano público (largo msecs) lanza InterruptedException {

if(Thread.interrupted())

lanza nueva InterruptedException;

sincronizado(this){

if(!flg){

flg=true;

devuelve verdadero

}si no (msecslt;=0){

devolver falso;

p>

}else{

long waitTime=msecs

long start=System.currentTimeMillis(); >

intentar{

for(;;){

esperar(waitTime);

if(!flg){

flg=verdadero

;

devuelve verdadero;

}else{

waitTime=msecs-(System.currentTimeMillis()-start);

si (waitTimelt;=0)

return false;

}

}

}catch(InterruptedException ex){

notificar();

lanzar ex

}

}

}

}

lanzamiento público vacío() {

flg=false;

notificar()

}

}