• 主页
  • 文章
  • 一般的
  • 什么是 Java 中的线程安全阻塞队列？ 你应该什么时候使用它？ 附执行

什么是 Java 中的线程安全阻塞队列？ 你应该什么时候使用它？ 附执行

已发表: 2021-04-30
什么是 Java 中的 Threadsafe BlockingQueue 以及何时应该使用它

到目前为止，我已经在 Crunchify 上写了两篇关于生产者消费者概念的文章。 第一个解释 Java Semaphore 和 Mutex，第二个解释并发读/写。

在本 Java 教程中，我们将通过相同的生产者/消费者概念来解释BlockingQueue in Java 。

Java中阻塞队列的优点是什么？

java.util.Queue支持在检索元素时等待队列变为非空，并在存储元素时等待队列中的空间可用的操作。

java中的java.util.BlockingQueue - Crunchify

我们需要创建四个 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 ，它创建了简单的 msg 并将其放入队列。

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

第三步

创建从队列中消费消息的类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 ( ) ;
         }
     }   
}

第四步

创建运行 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时，实现会抛出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 。 在这种情况下，最好向潜在的生产者发出队列已满的信号，并在失败时迅速放弃。
    • 换句话说：生产者自然受到限制。
  • 阻塞队列通常用于并发应用程序
  • 它提供了正确的、线程安全的实现
  • 内存消耗也应该受到限制