详解 JUC 之 BlockingQueue 接口以及 ArrayBlockingQueue 实现类

发表于 | 分类于 ， ， |
详解 JUC 之 BlockingQueue 接口以及 ArrayBlockingQueue 实现类

1 概述

  • java.util.concurrent.BlockingQueue

队列是一种 FIFO（先进先出）的数据结构，本文要讲的 BlockingQueue 也是一种队列，而且强调了线程安全的特性。

BlockingQueue 是一个线程安全的队列接口，多个线程能够以并发的方式从队列中插入数据，取出数据的同时不会出现线程安全的问题。

2 生产者和消费者例子

BlockingQueue 通常用于消费者线程向队列存入数据，消费者线程从队列中取出数据，具体如下

BlockingQueue

  1. 生产者线程不停的向队列中插入数据，直到队列满了，生产者线程被阻塞
  2. 消费者线程不停的从队列中取出数据，直到队列为空，消费者线程被阻塞

3 BlockingQueue 方法

BlockingQueue 提供 4 种不同类型的方法用于插入数，取出数据以及检查数据，具体如下

   1. 操作失败，抛出异常 2. 无论成功/失败，立即返回 true/false 3. 如果队列为空/满，阻塞当前线程 4. 如果队列为空/满，阻塞当前线程并有超时机制
插入 add(o) offer(o) put(o) offer(o, timeout, timeunit)
取出 remove(o) poll() take() poll(timeout, timeunit)
检查 element() peek()    

4 BlockingQueue 的具体实现类

BlockingQueue 只是一个接口，在实际开发中有如下的类实现了该接口。

  1. ArrayBlockingQueue
  2. DelayQueue
  3. LinkedBlockingQueue
  4. PriorityBlockingQueue
  5. SynchronousQueue

5 ArrayBlockingQueue 的使用

ArrayBlockingQueue

这里以 BlockingQueue 接口的具体实现类 ArrayBlockingQueue 举例。通过 ArrayBlockingQueue 实现一个消费者和生产者多线程模型。

核心内容如下

  1. 以 ArrayBlockingQueue 作为生产者和消费者的数据容器
  2. 通过 ExecutorService 启动 3 个线程，2 两个生产者，1 个消费者
  3. 指定数据总量

5.1 生产者线程

  • ArrayBlockingQueueProducer
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
42
43
44
45
46
47
48
49

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 生产者线程向容器存入指定总量的 任务
 *
 */
public class ArrayBlockingQueueProducer implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(ArrayBlockingQueueProducer.class);

    // 容器
    private ArrayBlockingQueue<String> queue;
    // 生产指定的数量
    private AtomicInteger numberOfElementsToProduce;

    public ArrayBlockingQueueProducer(ArrayBlockingQueue<String> queue, AtomicInteger numberOfElementsToProduce) {
        this.queue = queue;
        this.numberOfElementsToProduce = numberOfElementsToProduce;
    }

    @Override
    public void run() {
        try {
            while (numberOfElementsToProduce.get() > 0) {
                try {
                    // 向队列中存入任务
                    String task = String.format("task_%s", numberOfElementsToProduce.getAndUpdate(x -> x-1));
                    queue.put(task);
                    logger.info("thread {}, produce task {}", Thread.currentThread().getName(), task);

                    // 任务为0，生产者线程退出
                    if (numberOfElementsToProduce.get() == 0) {
                        break;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        } catch (Exception e) {
            logger.error(this.getClass().getName().concat(". has error"), e);
        }

    }
}

5.2 消费者线程

  • ArrayBlockingQueueConsumer
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
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 消费者线程向容器 消费 指定总量的任务
 *
 */
public class ArrayBlockingQueueConsumer implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(ArrayBlockingQueueConsumer.class);

    private ArrayBlockingQueue<String> queue;
    private AtomicInteger numberOfElementsToProduce;

    public ArrayBlockingQueueConsumer(ArrayBlockingQueue<String> queue, AtomicInteger numberOfElementsToProduce) {
        this.queue = queue;
        this.numberOfElementsToProduce = numberOfElementsToProduce;
    }

    @Override
    public void run() {
        try {
            while (!queue.isEmpty() || numberOfElementsToProduce.get() >= 0) {
                // 从队列中获取任务，并执行任务
                String task = queue.take();
                logger.info("thread {} consume task {}", Thread.currentThread().getName(),task);

                // 队列中数据为空，消费者线程退出
                if (queue.isEmpty()) {
                    break;
                }
            }
        } catch (Exception e) {
            logger.error(this.getClass().getName().concat(". has error"), e);
        }
    }
}

5.3 测试

  • TestBlockingQueue
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
import com.ckjava.synchronizeds.appCache.WaitUtils;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 1. 以 ArrayBlockingQueue 作为生产者和消费者的数据容器 <br>
 * 2. 通过 ExecutorService 启动 3 个线程，2 两个生产者，1 个消费者 <br>
 * 3. 指定数据总量
 */
public class TestBlockingQueue {
    public static void main(String[] args) {
        ArrayBlockingQueue<String> arrayBlockingQueue = new ArrayBlockingQueue<>(10);
        /*BlockingQueue delayQueue = new DelayQueue();
        BlockingQueue<String> linkedBlockingQueue = new LinkedBlockingQueue<>(10);
        BlockingQueue<String> priorityBlockingQueue = new PriorityBlockingQueue<>(10);
        BlockingQueue<String> synchronousQueue = new SynchronousQueue<>();*/

        ExecutorService executorService = Executors.newFixedThreadPool(3);
        // 最多生产 5 个数据
        AtomicInteger numberOfElementsToProduce = new AtomicInteger(5);

        // 2 个生产者线程
        executorService.submit(new ArrayBlockingQueueProducer(arrayBlockingQueue, numberOfElementsToProduce));
        executorService.submit(new ArrayBlockingQueueProducer(arrayBlockingQueue, numberOfElementsToProduce));
        // 1 个消费者线程
        executorService.submit(new ArrayBlockingQueueConsumer(arrayBlockingQueue, numberOfElementsToProduce));

        executorService.shutdown();
        WaitUtils.waitUntil(() -> executorService.isTerminated(), 1000L);
    }
}
  • 输出如下
1
2
3
4
5
6
7
8
9
10
13:54:17.884 [pool-1-thread-3] INFO  c.c.b.ArrayBlockingQueueConsumer - thread pool-1-thread-3 consume task task_5
13:54:17.884 [pool-1-thread-1] INFO  c.c.b.ArrayBlockingQueueProducer - thread pool-1-thread-1, produce task task_5
13:54:17.884 [pool-1-thread-2] INFO  c.c.b.ArrayBlockingQueueProducer - thread pool-1-thread-2, produce task task_4
13:54:17.887 [pool-1-thread-3] INFO  c.c.b.ArrayBlockingQueueConsumer - thread pool-1-thread-3 consume task task_4
13:54:17.887 [pool-1-thread-2] INFO  c.c.b.ArrayBlockingQueueProducer - thread pool-1-thread-2, produce task task_2
13:54:17.887 [pool-1-thread-1] INFO  c.c.b.ArrayBlockingQueueProducer - thread pool-1-thread-1, produce task task_3
13:54:17.887 [pool-1-thread-3] INFO  c.c.b.ArrayBlockingQueueConsumer - thread pool-1-thread-3 consume task task_3
13:54:17.887 [pool-1-thread-2] INFO  c.c.b.ArrayBlockingQueueProducer - thread pool-1-thread-2, produce task task_1
13:54:17.887 [pool-1-thread-3] INFO  c.c.b.ArrayBlockingQueueConsumer - thread pool-1-thread-3 consume task task_2
13:54:17.887 [pool-1-thread-3] INFO  c.c.b.ArrayBlockingQueueConsumer - thread pool-1-thread-3 consume task task_1

6 参考

Buy me a cup of coffee