【JavaEE】多线程(5) -- 阻塞队列

目录

1.阻塞队列是什么?

2.生产者消费者模型

3.标准库中的阻塞队列

4.阻塞队列的实现

---

1.阻塞队列是什么?

阻塞队列是?种特殊的队列. 也遵守 "先进先出" 的原则

阻塞队列能是?种线程安全的数据结构, 并且具有以下特性:

当队列满的时候, 继续?队列就会阻塞, 直到有其他线程从队列中取?元素.

?当队列空的时候, 继续出队列也会阻塞, 直到有其他线程往队列中插?元素.

阻塞队列的?个典型应?场景就是 "?产者消费者模型". 这是?种?常典型的开发模型.

2.生产者消费者模型

?产者消费者模式就是通过?个容器来解决?产者和消费者的强耦合问题。 ?产者和消费者彼此之间不直接通讯，?通过阻塞队列来进?通讯，所以?产者?产完数据之后不? 等待消费者处理，直接扔给阻塞队列，消费者不找?产者要数据，?是直接从阻塞队列?取.

实际开发中, 经常会涉及到 "分布式系统", 服务器整个功能不是由一个服务器全部完成的. 而是每个服务器负责一部分功能, 通过服务器之间的网络通信, 最终完成整个功能.

上述过程中, A 和 B , A 和 C 之间的耦合性是比较轻的, A中的代码需要设计到一些B相关的操作, B中的代码也涉及到和A相关的操作,? ?A 的代码中也需要涉及和 C 相关的操作, C 的代码也涉及和 A 相关的操作, 另外, 如果 B 或者 C 服务器出现故障, 对 A 的影响就很大.

引用生产者消费者模型, 就可以降低上述的耦合

这样 A B C 之间就不是直接交互了, 而是通过队列在中间经常传递.

3.标准库中的阻塞队列

使用实例:

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class ThreadDemo28 {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(100);
        queue.put("hello");
        System.out.println(queue.take());
        System.out.println(queue.take());

    }
}

当队列中的元素为空时, 在执行 take() 方法, 进程就会进入阻塞状态:

此时进程并会不结束.

4.阻塞队列的实现

基于环形队列来简单实现阻塞队列:

import static java.lang.Thread.sleep;

class MyBlockingQueue {
    private String[] elems = null;
    private int head = 0;
    private int tail = 0;
    private int size = 0;
    Object locker = new Object();

    public MyBlockingQueue(int capacity) {
        elems = new String[capacity];
    }

    public void put(String elem) throws InterruptedException {
        synchronized (locker) {
            while(size >= elem.length()) {
                //队列满了就阻塞
                locker.wait();
            }

            elems[tail] = elem;
            tail++;
            if(tail >= elems.length) {
                tail = 0;
            }
            size++;

            // 入队列成功后唤醒
            locker.notify();
        }
    }

    public String take() throws InterruptedException {
        String elem = null;
        synchronized (locker) {
            while(size == 0) {
                // 队列空了
                // 也需要这个代码阻塞
                locker.wait();
            }

            elem = elems[head];
            head++;
            if(head >= elems.length) {
                head = 0;
            }
            size--;

            //元素出队列成功后, 唤醒
            locker.notify();
        }

        return elem;
    }
}

细节注意:

基于阻塞队列实现简单的生产者消费者模型:

public class ThreadDeom29 {
    public static void main(String[] args) {
        MyBlockingQueue queue = new MyBlockingQueue(1000);
        //生产者
        Thread t1 = new Thread(()->{
            int n = 1;
            while(true) {
                try {
                    queue.put(n + "");
                    System.out.println("生产元素 " + n);
                    sleep(1000);
                    n++;
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });

        //消费者
        Thread t2 = new Thread(()-> {
            while(true) {
                try {
                    String n = queue.take();
                    System.out.println("消费元素 " + n);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });

        t1.start();
        t2.start();
    }
}

t1线程每隔一秒钟存放一个元素到阻塞队列中, t2 线程则在队列中有元素时将元素取出, 队列没有元素就进入阻塞状态, 直到队列不再为空?

?