Java多线程案例之阻塞队列

愿荣 2023-10-22 原文

文章目录

一. 认识阻塞队列

1. 什么是阻塞队列

阻塞队列本质上还是一种队列, 和普通队列一样, 遵循先进先出, 后进后出的规则, 但阻塞队例相比于普通队列的特殊之处在于阻塞队列的阻塞功能, 主要基于多线程使用.

如果队列为空, 执行出队列操作, 就会使线程陷入阻塞, 阻塞到另一个线程往队列里添加元素(队列不空)为止.
如果队列满了，执行入队列操作, 也会使线程阻塞, 阻塞到另一个线程从队列取走元素位置(队列不满)为止.

2. 生产者消费者模型

基于阻塞队列的阻塞特性是可以实现 “生产者消费者模型” 的, 那么什么是生产者消费者模型呢?

还是用生活中的例子来解释, 这不还有半个月就要过年了, 大年当晚我们会吃年夜饭, 饺子就是年夜饭当中的一份主食, 那么要想吃到饺子, 最好就是一家人在一起把饺子包好, 简单来讲包饺子的步骤有: 擀饺子皮＋包饺子.

有下面两种包饺子方式 :

每个人, 都分别进行擀饺子皮＋包饺子这样的操作, 但毕竟家里面的擀面杖不会准备那么多吧, 大家会竞争面杖, 一个人使用擀面杖的使用, 其他人就得阻塞等待, 这就影响了包饺子的效率了.
一个人专门负责擀饺子皮, 另外三个人负责包, 擀饺子的人每次擀好一个皮, 就放到盖帘上, 其他人每次都从盖帘上取一个皮包饺子.

其实第二种包饺子方式就是生产者消费者模型的运用, 擀饺子皮的那个人就是生产者, 其他负责包饺子的人就是消费者, 放饺子皮的盖帘就相当于阻塞队列, 如果擀饺子皮的人擀的太慢生产的饺子皮供不上使用, 不一会盖帘上没皮了, 包饺子的人就得等一会儿再包; 如果擀饺子的人擀的太快了, 包的速度跟不上擀的速度, 盖帘上放满了饺子皮, 擀饺子皮的人就得等一会儿再擀.

生产者消费者模型能够给程序带来两个非常重要的好处, 一是可以实现实现了发送方和接收方之间的 “解耦” , 二是可以 “削峰填谷” , 保证系统的稳定性, 具体理解如下:

在服务器相互调用的场景中假设有两个服务器A(请求服务器), B(应用服务器), A把请求转发给B处理, B处理完了把结果反馈给A, 这种情况下, A和B之间的耦合是比较高的, A要调用B, A 务必要知道B的存在, 如果B挂了, 很容易引起A的bug, 再比如再加一个C服务器, 此时也需要对A修改不少代码, 就需要针对A重新修改代码, 重新测试, 重新发布, 重新部署等, 这就非常麻烦了.

而针对上述场景, 使用生产者消费者模型就可以有效的降低耦合,

A和B之间通过一个阻塞队列来通信, 此时A是不知道B的, A只知道队列, 也就是说A的代码中没有任何一行代码和B相关; 同样的, B也是不知道A的, B也是只知道队列, B的代码中,也没有任何一行代码和A相关.

如果B挂了, 对于A没有任何影响, 因为队列还是正常的, A仍然可以给队列插入元素, 如果队列满就先阻塞等待; 同样如果A挂了, 也对于B没啥影响, 队列是正常的B就仍然可以从队列取元素, 如果队列空了, 也就阻塞等待就好了; 也就是说, AB任何一方挂了不会对对方造成影响, 同时, 新增一个C来作为消费者, 对于A来说仍然是无感知的.

“削峰填谷” 可以联想三峡大坝的水库, 三峡大坝的水库,

如果上游水多了, 三峡大坝就会关闸蓄水, 此时就相当于由三峡大坝承担了上游的冲击, 对下游起到了很好的保护左右, 这就是 “削峰” 作用; 如果上游水少了, 三峡大坝开闸放水, 有效保证下游的用水情况, 避免出现干旱灾害, 这就是 “填谷” 作用.

而在服务器开发中, 上游就是用户发送的请求, 下游就是一些执行具体业务的服务器, 用户发多少请求这是不可控的, 有的时候请求多, 有的时候请求少, 而如果没有使用生产者消费者模型, A服务器用户请求暴涨, 此时如果没有充分的准备, B服务器来不及响应一下没抗住, 就可能会挂掉.

但是, 如果使用生产者消费者模型, 那么即使A服务器请求暴涨, 也不会影响到B, 这是因为A请求暴涨后, 用户的请求都被打包到阻塞队列中(如果阻塞队列有界, 则会引起队列阻塞, 不会影响到B), B可以从阻塞队列中取出元素以合适的速度来处理这些请求, 这就是 “削峰填谷” 的作用了.

3. 标准库中阻塞队列类

Java标准库也提供了阻塞队列的标准类, 常用的有下面几个：

ArrayBlockingQueue : 基于数组实现界阻塞队列
LinkedBlockingQueue : 基于链表实现的有界阻塞队列
PriorityBlockingQueue : 带有优先级(堆)的无界阻塞队列
BlockingQueue接口 : 上面的类实现了该接口

阻塞队列类的核心方法：

方法	解释
void put(E e) throws InterruptedException	带有阻塞特性的入队操作方法
E take() throws InterruptedException	带有阻塞特性的出队操作方法
boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException	带有阻塞特性的入队操作方法, 并且可以设置最长等待时间
E poll(long timeout, TimeUnit unit) throws InterruptedException	带有阻塞特性的出队操作方法, 并且可以设置最长等待时间

注意 : 其他一些重载的offer, poll等普通队列方法也是支持使用的, 但是这些方法就不带有阻塞功能了.

代码示例:

下面的代码是基于标准库的阻塞队列简单实现的生产者消费者模型.

public class TestDemo19 {
    public static void main(String[] args) {
        BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>();
        //消费者线程
        Thread customer = new Thread(() -> {
           while (true) {
               try {
                   Integer result = blockingQueue.take();
                   System.out.println("消费元素: " + result);
               } catch (InterruptedException e) {
                   throw new RuntimeException(e);
               }
           }
        });
        customer.start();

        //生产者线程
        Thread producer = new Thread(() -> {
            int count = 0;
            while (true) {
                try {
                    blockingQueue.put(count);
                    System.out.println("生产元素: " + count);
                    count++;
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        producer.start();
    }
}

执行结果:

二. 基于循环队列实现的简单阻塞队列

1. 循环队列的简单实现

要实现一个阻塞队列, 需要先实现一个普通的循环队列, 循环队列是基于数组实现的, 这里最重要的是如何将队列为空状态与满状态区分开来, 对于这里的实现不懂得可以看看看我前面博客中关于循环队列的内容, 这里就是不做过多的赘述了 : 队列与集合Queue,Deque的理解和使用 , 用栈实现队列,用队列实现栈,设计循环队列 , 阻塞队列最核心的就是出队和入队操作, 所以我们这里重点实现这两个方法, 代码如下:

//普通的循环队列
class MyBlockingQueue {
    //存放元素的数数组
    private int[] items = new int[1000];
    //队头指针
    private int head = 0;
    //队尾指针
    private int tail = 0;
    //记录队列元素的个数
    private int size = 0;

    //入队操作
    public void put (int val) {
        if (size == items.length) {
            //队列满了
            return;
        }
        items[tail++] = val;
        //等价于 tail %= items.length
        if (tail >= items.length) {
            tail = 0;
        }
        size++;
    }

    //出队操作
    public Integer take() {
        int resulet = 0;
        if (size == 0) {
            //队列空了
            return null;
        }
        resulet = items[head++];
        //等价于 head %= elem.length
        if (head >= items.length) {
            head = 0;
        }
        size--;
        return resulet;
    }
}

2. 阻塞队列的简单实现

首先要实现的阻塞队列在大多数情况下是在多线程情况下使用的, 所以要考虑线程安全问题, 上面循环队列的代码take与put方法都有写操作, 直接加锁即可.

//线程安全的循环队列
class MyBlockingQueue {
    //存放元素的数数组
    private int[] items = new int[1000];
    //队头指针
    private int head = 0;
    //队尾指针
    private int tail = 0;
    //记录队列元素的个数
    private int size = 0;

    //入队操作
    public void put (int val) {
        synchronized (this) {
            if (size == items.length) {
                //队列满了
                return;
            }
            items[tail++] = val;
            //等价于 tail %= items.length
            if (tail >= items.length) {
                tail = 0;
            }
            size++;
        }
    }

    //出队操作
    public Integer take() {
        int resulet = 0;
        synchronized (this) {
            if (size == 0) {
                //队列空了
                return null;
            }
            resulet = items[head++];
            //等价于 head %= elem.length
            if (head >= items.length) {
                head = 0;
            }
            size--;
            return resulet;
        }
    }
}

然后就是实现阻塞效果了, 主要是使用wait和notify实现线程的阻塞等待.

入队时, 队列满了需要使用wait方法使线程阻塞, 直到有元素出队队列不满了再使用notify通知线程执行.

出队时, 队列为空也需要使用wait方法使线程阻塞, 直到有新元素入队再使用notify通知线程执行.

代码如下:

class MyBlockingQueue {
    //存放元素的数数组
    private int[] items = new int[1000];
    //队头指针
    private int head = 0;
    //队尾指针
    private int tail = 0;
    //记录队列元素的个数
    private int size = 0;

    //入队操作
    public void put (int val) throws InterruptedException {
        synchronized (this) {
            if (size == items.length) {
                //队列满了,阻塞等待
                this.wait();
            }
            items[tail++] = val;
            //等价于 tail %= items.length
            if (tail >= items.length) {
                tail = 0;
            }
            size++;
            //唤醒因队列空造成的阻塞wait
            this.notify();
        }
    }

    //出队操作
    public Integer take() throws InterruptedException {
        int resulet = 0;
        synchronized (this) {
            if (size == 0) {
                //队列空了,阻塞等待
                this.wait();
            }
            resulet = items[head++];
            //等价于 head %= elem.length
            if (head >= items.length) {
                head = 0;
            }
            size--;
            //唤醒因队列满造成的阻塞wait
            this.notify();
            return resulet;
        }
    }
}

上述代码已经基本实现了阻塞队列的功能, 但不够完善, 这里的代码再改良一下把判断队列满或者空的wait部分的代码, 把if改成while是更好的, 为什么这样写呢?

我们思考当代码中当wait被唤醒的时候, 此时if的条件一定就不成立了吗?

具体来说, 思考put方法中的wait被唤醒, 往下执行是要要求，队列不满但是wait被唤醒了之后, 队列一定是不满的嘛? 其实当前代码中是不会出现这样的问题的, 但是稳妥起见, 最好的办法就是wait唤醒之后再判定一下条件是否满足, 而且Java标准库当中就是建议这么写的.

调整部分的代码如下:

//出队部分
while (size == items.length) {
    //队列满了,阻塞等待
    this.wait();
}

//入队部分
while (size == 0) {
    //队列空了,阻塞等待
    this.wait();
}

最后就是测试一下我们所写的这个阻塞队列了, 我们创建两个线程分别是消费者线程customer和生产者线程producer, 生产者生产数字, 消费者消费数字, 为了让执行结果中的阻塞效果明显一些, 我们可以使用sleep方法来控制一下生产者/消费者的生产/消费的频率, 这里我们让开始时生产的速度快一些, 消费的速度慢一些, 全部代码如下:

class MyBlockingQueue {
    //存放元素的数数组
    private int[] items = new int[1000];
    //队头指针
    private int head = 0;
    //队尾指针
    private int tail = 0;
    //记录队列元素的个数
    private int size = 0;

    //入队操作
    public void put (int val) throws InterruptedException {
        synchronized (this) {
            while (size == items.length) {
                //队列满了,阻塞等待
                this.wait();
            }
            items[tail++] = val;
            //等价于 tail %= items.length
            if (tail >= items.length) {
                tail = 0;
            }
            size++;
            //唤醒因队列空造成的阻塞wait
            this.notify();
        }
    }

    //出队操作
    public Integer take() throws InterruptedException {
        int resulet = 0;
        synchronized (this) {
            while (size == 0) {
                //队列空了,阻塞等待
                this.wait();
            }
            resulet = items[head++];
            //等价于 head %= elem.length
            while (head >= items.length) {
                head = 0;
            }
            size--;
            //唤醒因队列满造成的阻塞wait
            this.notify();
            return resulet;
        }
    }
}

public class Test {
    public static void main(String[] args) {
        //消费线程
        MyBlockingQueue queue = new MyBlockingQueue();
        Thread customer = new Thread(() -> {
            while(true) {
                try {
                    int result = queue.take();
                    System.out.println("消费元素: " + result);
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        customer.start();
        //生产线程
        Thread producer = new Thread(() -> {
            int count = 0;
            while (true) {
                try {
                    queue.put(count);
                    System.out.println("生产元素: " + count);
                    count++;
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        producer.start();
    }
}