[JAVAee]阻塞队列

阻塞队列的含义

有队列这两个字的,少不了"先进先出"这个特性

阻塞队列是一种线程安全的数据结构,主要的特性有:

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

阻塞队列的使用

在java标准库中,内置了阻塞队列.当我们想使用的时候,可以实现BlockingQueue接口就好(但其真正实现的类为LinkedBlockingQueue).

在队列当中,必然有方法将元素进行放置或取出.

在阻塞队列中,常用的为put方法放置元素与take方法取出元素,也只有这两种方法具有阻塞的特性.能实现阻塞队列的效果.

public static void main(String[] args) throws InterruptedException {BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();queue.put(5);//放置一个元素,如果当前队列已经满了则此线程进入阻塞状态queue.take();//取出一个元素,如果当前队列为空则此线程进入阻塞状态}

阻塞队列的模拟实现

实现阻塞队列:

实现一个循环队列()
给放置与取出方法加上阻塞功能
保证线程安全

①实现循环队列,与普通的put与take方法

class MyBlockingQueue{private int[] array = new int[1000];//基于数组来实现队列private int size = 0;//记录队列当前的元素个数private int head = 0;//记录头节点private int tail = 0;//记录尾巴节点public void put(int value){if(size == array.length){//检测队列中的元素是否已满System.out.println("已满,勿扰");return;}array[tail++] = value;//尾插法size++;if(tail == array.length){tail = 0;//达到循环队列的效果}}public int take(){if(size == 0){System.out.println("空了空了");return -1;}int ret = array[head];head++;if(head == array.length){head = 0;//循环循环}size--;return ret;}
}

②添加阻塞功能

分别往put与take方法加上wait与notifyAll

put时队列满的时候进入阻塞状态,只有等到调用take方法取走一个元素后才会唤醒线程
take时队列空的时候进入阻塞状态,只有等到调用put方法放置一个元素后才会唤醒线程

public void put(int value) throws InterruptedException {if(size == array.length){//检测队列中的元素是否已满wait();}array[tail++] = value;//尾插法size++;if(tail == array.length){tail = 0;//达到循环队列的效果}notifyAll();}

public int take()throws InterruptedException{if(size == 0){//检测队列是否为空wait();}int ret = array[head];head++;if(head == array.length){head = 0;//循环循环}size--;notifyAll();return ret;}

③保证线程安全

关于线程安全要注意的是:

多线程环境下的抢占式执行
多个线程修改同一个变量
线程的读写操作不能保证原子性

对于此处的读写修改操作,emmm.

put方法和take方法好像一整个都是呢

所以我们可以使用synchronized关键字为他们上锁,保证原子性达到线程安全.

对于频繁修改和需要进行比较或判断的元素,也要加上volatile关键字,保证线程内存可见性.防止存在误判导致的一系列错误.

class MyBlockingQueue{private int[] array = new int[1000];//基于数组来实现队列private volatile int size = 0;//记录队列当前的元素个数private int head = 0;//记录头节点private int tail = 0;//记录尾巴节点public void put(int value) throws InterruptedException {synchronized (this) {while(size == array.length){//使用while循环,线程被唤醒后继续判断队列是否已满//真正使用的情况下不但只有两个线程//检测队列中的元素是否已满wait();}array[tail++] = value;//尾插法size++;if(tail == array.length){tail = 0;//达到循环队列的效果}notifyAll();}}public int take()throws InterruptedException{synchronized (this) {while(size == 0){//检测队列是否为空wait();}int ret = array[head];head++;if(head == array.length){head = 0;//循环循环}size--;notifyAll();return ret;}}
}