Qu'est-ce que Thread-Safe BlockingQueue en Java ? Quand faut-il l'utiliser ? Mise en œuvre jointe

Publié: 2021-04-30
Qu'est-ce que Threadsafe BlockingQueue en Java et quand l'utiliser

Jusqu'à présent, j'ai écrit deux articles sur le concept Producteur Consommateur sur Crunchify. Le 1er pour expliquer Java Semaphore et Mutex et le 2ème pour expliquer Concurrent Read/Write.

Dans ce didacticiel Java, nous passerons en revue le même concept Producteur/Consommateur pour expliquer la BlockingQueue in Java .

Quels sont les avantages de Blocking Queue en Java ?

Une file d'attente java.util.Queue prend en charge les opérations qui attendent que la file d'attente ne soit plus vide lors de la récupération d'un élément et attendent que de l'espace se libère dans la file d'attente lors du stockage d'un élément.

java.util.BlockingQueue en java - Crunchify

Nous devons créer quatre classes Java :

  1. CrunchifyMessage.java pour mettre et recevoir un message
  2. CrunchifyBlockingProducer.java pour mettre le message dans la file d'attente
  3. CrunchifyBlockingConsumer.java pour obtenir le message de la file d'attente
  4. CrunchifyBlockingMain.java pour démarrer le test

Les implémentations de BlockingQueue sont thread-safe . Toutes les méthodes de mise en file d'attente sont de nature atomique et utilisent des verrous internes.

Commençons par l'implémentation de Thread-Safe BlockingQueue en Java

Étape 1

Créez la classe CrunchifyMessage.java . Il s'agit d'un simple objet Java.

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 ;
     }
}

Étape 2

Créez le producteur CrunchifyBlockingProducer.java qui a créé un message simple et l'a mis en file d'attente.

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 ) ;
         }
     }   
}

Étape 3

Créez la classe CrunchifyBlockingConsumer.java qui consomme les messages de la file d'attente.

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 ( ) ;
         }
     }   
}

Étape 4

Créez une méthode CrunchifyBlockingMain.java simple qui exécute le test BlockingQueue. Exécutez ce programme pour vérifier le comportement 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" ) ;
     }   
}

Une BlockingQueue n'accepte pas les éléments nuls . Les implémentations lancent NullPointerException sur les tentatives d' ajout , de placement ou d'offre d' un null .

Une valeur nulle est utilisée comme valeur sentinelle pour indiquer l'échec des opérations d' interrogation .

Résultat:

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 .

Quand devrions-nous utiliser java.util.concurrent.BlockingQueue ?

  • Lorsque vous souhaitez limiter une sorte de demande entrante, vous devez utiliser le même
  • Un producteur peut prendre une longueur d'avance sur les consommateurs avec une file d'attente illimitée. Si le consommateur ne rattrape pas le producteur, cela peut provoquer une OutOfMemoryError . Dans de telles situations, il peut être préférable de signaler à un producteur potentiel que la file d'attente est pleine et d'abandonner rapidement en cas d'échec.
    • Autrement dit : les producteurs sont naturellement étranglés.
  • La file d'attente de blocage est normalement utilisée dans les applications simultanées
  • Il fournit une implémentation correcte et thread-safe
  • La consommation de mémoire doit également être limitée