Java 阻塞队列实现生产者和消费者场景_对于生产者-消费者并发场景,建议使用阻塞队列

1. 阻塞队列

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作是：

• 在队列为空时，获取元素的线程会等待队列变为非空。
• 当队列满时，存储元素的线程会等待队列可用。

当阻塞队列满时，如果生产者线程使用put方法向队列中添加元素，队列会一直阻塞生产者线程，直到拿到数据，或者响应中断退出。当队列空时，消费者使用take方法从队列中获取元素，队列也会阻塞消费者线程，直到队列可用。

2. 阻塞队列实现生产者和消费者实例

生产者线程，ProducerThread.java:

import java.util.Random;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 生产者线程
 * @author syrdbt
 * @date 2020-02-26
 */
public class ProducerThread implements Runnable{

    private LinkedBlockingQueue blockingQueue;

    public ProducerThread(LinkedBlockingQueue blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    @Override
    public void run() {
        Random random = new Random();
        while (true) {
            try {
                this.produce(random.nextInt(5));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private void produce(int productNumber) throws InterruptedException {
        TimeUnit.SECONDS.sleep(productNumber);
        blockingQueue.put(productNumber);
        System.out.println("【生产者】生产了" + productNumber + "号产品，花费了" + productNumber + "秒，成功的添加"
                + productNumber + "产品到了队列中");
    }
}

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

消费者线程，ConsumerThread.java：

/**
 * 消费者线程
 * @author syrdbt
 * @date 2020-02-26
 */
public class ConsumerThread implements Runnable{

    private LinkedBlockingQueue blockingQueue;

    public ConsumerThread(LinkedBlockingQueue blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    @Override
    public void run() {
        Random random = new Random();
        while (true) {
            try {
                this.consume();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private void consume() throws InterruptedException {
        int productNumber = (int) blockingQueue.take();
        TimeUnit.SECONDS.sleep(productNumber);
        System.out.println("【消费者】消费了" + productNumber + "号产品，花费了" +
                productNumber + "秒，成功的从队列中取出" + productNumber + "号产品");
    }
}
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

测试程序，Main.java：

import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author syrdbt
 * @date 2020-02-26
 */
public class Main {
    public static void main(String[] args) {
        LinkedBlockingQueue blockingQueue = new LinkedBlockingQueue();
        Thread producerThread = new Thread(new ProducerThread(blockingQueue));
        Thread consumerThread = new Thread(new ConsumerThread(blockingQueue));
        producerThread.start();
        consumerThread.start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

测试程序运行截图如下：

参考文献

• 聊聊并发（七）——Java中的阻塞队列