¿Qué es BlockingQueue seguro para subprocesos en Java? ¿Cuándo debes usarlo? Implementación adjunta

Publicado: 2021-04-30
¿Qué es Threadsafe BlockingQueue en Java y cuándo debería usarlo?

Hasta ahora he escrito dos artículos sobre el concepto Producer Consumer en Crunchify. El primero para explicar Java Semaphore y Mutex y el segundo para explicar la lectura/escritura simultánea.

En este tutorial de Java, repasaremos el mismo concepto de productor/consumidor para explicar BlockingQueue in Java .

¿Cuáles son las ventajas de Blocking Queue en Java?

Una java.util.Queue admite operaciones que esperan a que la cola deje de estar vacía al recuperar un elemento y espera a que haya espacio disponible en la cola al almacenar un elemento.

java.util.BlockingQueue en java - Crunchify

Necesitamos crear cuatro clases de Java:

  1. CrunchifyMessage.java para poner y recibir mensajes
  2. CrunchifyBlockingProducer.java para poner el mensaje en cola
  3. CrunchifyBlockingConsumer.java para obtener el mensaje de la cola
  4. CrunchifyBlockingMain.java para iniciar la prueba

Las implementaciones de BlockingQueue son thread-safe . Todos los métodos de cola son de naturaleza atómica y utilizan bloqueos internos.

Comencemos con la implementación de Thread-Safe BlockingQueue en Java

Paso 1

Crea la clase CrunchifyMessage.java . Este es un objeto Java simple.

Java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
package com . crunchify . example ;
 
/**
* @author Crunchify.com
* simple Message class to put and get message into queue
*/
 
public class CrunchifyMessage {
     private String crunchifyMsg ;
    
     public CrunchifyMessage ( String string ) {
         this . crunchifyMsg = string ;
     }
    
     public String getMsg ( ) {
         return crunchifyMsg ;
     }
}

Paso 2

Cree el productor CrunchifyBlockingProducer.java que creó un mensaje simple y lo puso en cola.

Java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
package com . crunchify . example ;
 
import java . util . concurrent . BlockingQueue ;
 
/**
* @author Crunchify.com
*
*/
 
public class CrunchifyBlockingProducer implements Runnable {
    
     private BlockingQueue <CrunchifyMessage> crunchQueue ;
    
     public CrunchifyBlockingProducer ( BlockingQueue <CrunchifyMessage> queue ) {
         this . crunchQueue = queue ;
     }
    
     @Override
     public void run ( ) {
         // producing CrunchifyMessage messages
         for ( int i = 1 ; i < = 5 ; i ++ ) {
             CrunchifyMessage msg = new CrunchifyMessage ( "i'm msg " + i ) ;
             try {
                 Thread . sleep ( 10 ) ;
                 crunchQueue . put ( msg ) ;
                 System . out . println ( "CrunchifyBlockingProducer: Message - " + msg . getMsg ( ) + " produced." ) ;
             } catch ( Exception e ) {
                 System . out . println ( "Exception:" + e ) ;
             }
         }
        
         // adding exit message
         CrunchifyMessage msg = new CrunchifyMessage ( "All done from Producer side. Produced 50 CrunchifyMessages" ) ;
         try {
             crunchQueue . put ( msg ) ;
             System . out . println ( "CrunchifyBlockingProducer: Exit Message - " + msg . getMsg ( ) ) ;
         } catch ( Exception e ) {
             System . out . println ( "Exception:" + e ) ;
         }
     }   
}

Paso 3

Cree la clase CrunchifyBlockingConsumer.java que consume el mensaje de la cola.

Java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
package com . crunchify . example ;
 
import java . util . concurrent . BlockingQueue ;
 
/**
* @author Crunchify.com
*
*/
 
public class CrunchifyBlockingConsumer implements Runnable {
    
     private BlockingQueue <CrunchifyMessage> queue ;
    
     public CrunchifyBlockingConsumer ( BlockingQueue <CrunchifyMessage> queue ) {
         this . queue = queue ;
     }
    
     @Override
     public void run ( ) {
         try {
             CrunchifyMessage msg ;
            
             // consuming messages until exit message is received
             while ( ( msg = queue . take ( ) ) . getMsg ( ) ! = "exit" ) {
                 Thread . sleep ( 10 ) ;
                 System . out . println ( "CrunchifyBlockingConsumer: Message - " + msg . getMsg ( ) + " consumed." ) ;
             }
         } catch ( InterruptedException e ) {
             e . printStackTrace ( ) ;
         }
     }   
}

Etapa 4

Cree el método simple CrunchifyBlockingMain.java que ejecuta la prueba BlockingQueue. Ejecute este programa para verificar el comportamiento de BlockingQueue.

Java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
package com . crunchify . example ;
 
import java . util . concurrent . ArrayBlockingQueue ;
import java . util . concurrent . BlockingQueue ;
 
/**
* @author Crunchify.com
*
*/
 
public class CrunchifyBlockingMain {
    
     public static void main ( String [ ] args ) {
        
         // Creating BlockingQueue of size 10
         // BlockingQueue supports operations that wait for the queue to become non-empty when retrieving an element, and
         // wait for space to become available in the queue when storing an element.
         BlockingQueue <CrunchifyMessage> crunchQueue = new ArrayBlockingQueue < > ( 10 ) ;
         CrunchifyBlockingProducer crunchProducer = new CrunchifyBlockingProducer ( crunchQueue ) ;
         CrunchifyBlockingConsumer crunchConsumer = new CrunchifyBlockingConsumer ( crunchQueue ) ;
        
         // starting producer to produce messages in queue
         new Thread ( crunchProducer ) . start ( ) ;
        
         // starting consumer to consume messages from queue
         new Thread ( crunchConsumer ) . start ( ) ;
        
         System . out . println ( "Let's get started. Producer / Consumer Test Started.\n" ) ;
     }   
}

Una BlockingQueue no acepta elementos nulos . Las implementaciones lanzan NullPointerException en los intentos de agregar , poner u ofrecer un valor nulo .

Un valor nulo se usa como valor centinela para indicar fallas en las operaciones de sondeo .

Resultado:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
Let 's get started. Producer / Consumer Test Started.
 
CrunchifyBlockingProducer: Message - i' m msg 1 produced .
CrunchifyBlockingProducer : Message - i 'm msg 2 produced.
CrunchifyBlockingConsumer: Message - i' m msg 1 consumed .
CrunchifyBlockingConsumer : Message - i 'm msg 2 consumed.
CrunchifyBlockingProducer: Message - i' m msg 3 produced .
CrunchifyBlockingConsumer : Message - i 'm msg 3 consumed.
CrunchifyBlockingProducer: Message - i' m msg 4 produced .
CrunchifyBlockingConsumer : Message - i 'm msg 4 consumed.
CrunchifyBlockingProducer: Message - i' m msg 5 produced .
CrunchifyBlockingProducer : Exit Message - All done from Producer side . Produced 50 CrunchifyMessages
CrunchifyBlockingConsumer : Message - i ' m msg 5 consumed .
CrunchifyBlockingConsumer : Message - All done from Producer side . Produced 50 CrunchifyMessages consumed .

¿Cuándo debemos usar java.util.concurrent.BlockingQueue?

  • Cuando desee acelerar algún tipo de solicitud entrante, debe usar el mismo
  • Los productores pueden adelantarse mucho a los consumidores con una cola ilimitada. Si el consumidor no se está poniendo al día con el productor, puede causar un OutOfMemoryError . En situaciones como estas, puede ser mejor señalar a un posible productor que la cola está llena y darse por vencido rápidamente si falla.
    • En otras palabras: los productores están naturalmente estrangulados.
  • La cola de bloqueo se usa normalmente en aplicaciones concurrentes
  • Proporciona una implementación correcta y segura para subprocesos.
  • El consumo de memoria también debe ser limitado