JavaのスレッドセーフBlockingQueueとは何ですか？ いつ使うべきですか？ 実装が添付されています

公開: 2021-04-30
JavaのThreadsafeBlockingQueueとは何ですか、いつ使用する必要がありますか

これまで、Crunchifyのプロデューサーコンシューマーの概念に関する2つの記事を書きました。 1つ目はJavaセマフォとMutexについて説明し、2つ目は同時読み取り/書き込みについて説明します。

このJavaチュートリアルでは、同じプロデューサー/コンシューマーの概念をBlockingQueue in Javaについて説明します。

Javaでキューをブロックする利点は何ですか？

java.util.Queueは、要素を取得するときにキューが空でなくなるのを待ち、要素を格納するときにキューでスペースが使用可能になるのを待つ操作をサポートします。

javaのjava.util.BlockingQueue-Crunchify

4つのJavaクラスを作成する必要があります。

  1. メッセージを入れて取得するためのCrunchifyMessage.java
  2. CrunchifyBlockingProducer.javaを使用して、メッセージをキューに入れます
  3. キューからメッセージを取得するためのCrunchifyBlockingConsumer.java
  4. CrunchifyBlockingMain.javaでテストを開始します

BlockingQueue thread-safeです。 すべてのキューイングメソッドは本質的にアトミックであり、内部ロックを使用します。

JavaでのスレッドセーフなBlockingQueueの実装を始めましょう

ステップ1

クラスCrunchifyMessage.javaを作成します。 これは単純な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 ;
     }
}

ステップ2

単純なメッセージを作成してキューに入れるプロデューサーCrunchifyBlockingProducer.javaを作成します。

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

ステップ-3

キューからのメッセージを消費するクラスCrunchifyBlockingConsumer.javaを作成します。

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

ステップ-4

BlockingQueueテストを実行する単純なCrunchifyBlockingMain.javaメソッドを作成します。 このプログラムを実行して、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" ) ;
     }   
}

BlockingQueueはnull要素を受け入れません。 実装は、 nullを追加、配置、または提供しようとするとNullPointerExceptionをスローします。

nullは、ポーリング操作の失敗を示す番兵値として使用されます。

結果：

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 .

java.util.concurrent.BlockingQueueをいつ使用する必要がありますか？

  • ある種の着信要求を抑制したい場合は、同じものを使用する必要があります
  • プロデューサーは、無制限のキューでコンシューマーよりもはるかに先を行くことができます。 コンシューマーがプロデューサーに追いついていない場合は、 OutOfMemoryErrorが発生する可能性があります。 このような状況では、キューがいっぱいであることをプロデューサー候補に通知し、失敗した場合はすぐに諦める方がよい場合があります。
    • 言い換えれば、生産者は自然に抑制されます。
  • ブロッキングキューは通常、並行アプリケーションで使用されます
  • 正しく、スレッドセーフな実装を提供します
  • メモリ消費も制限する必要があります