大型网站建设公司,便宜做网站怎么样,wordpress相关问题,北京网站平台开发文章目录 #x1f4c4;前言一. 阻塞队列初了解#x1f346;1. 什么是阻塞队列#xff1f;#x1f345;2. 为什么使用阻塞队列#xff1f;#x1f966;3. Java标准库中阻塞队列的实现 二. 阻塞队列的模拟实现#x1f35a;1. 实现普通队列#x1f365;2. 实现队列的阻塞功… 文章目录 前言一. 阻塞队列初了解1. 什么是阻塞队列2. 为什么使用阻塞队列3. Java标准库中阻塞队列的实现 二. 阻塞队列的模拟实现1. 实现普通队列2. 实现队列的阻塞功能3. 解除阻塞状态 三. 使用模拟的阻塞队列验证生产者和消费者模型 前言
本文是对阻塞队列的应用场景的介绍对阻塞队列的作用以及具体实现的讨论。 一. 阻塞队列初了解
1. 什么是阻塞队列
阻塞队列是一种带有阻塞功能的“先进先出”线性表。即在一个带有最大容量的队列中在某时刻队列容量已满时继续入队 或 队列为空时继续出队就会进入阻塞等待状态直到队列变为 未满或非空 便解除阻塞状态继续入队或出队。
2. 为什么使用阻塞队列
若存在以下简易的分布式系统 上述分布式系统虽然能完成客户端与服务器端的交互需求但可能存在以下问题
在正常情况下用户可以通过客户端想服务器发起请求并获取相应的服务但假如在某刻服务器A突然出现了故障与服务器A直接通信的服务器B也可能因此出现故障导致整个服务瘫痪。若未来想增加 更多的服务器 来处理服务器A发起的请求则需求对 服务器A 的接口 进行一定的改动付出一定的时间和人力成本。当某个时刻很多的客户端同时向 服务器A 发起请求作为与用户直接交互的服务器服务器A具备承载这些并发量的能力但服务器集群中负责其他功能的服务器接收请求的承载能力可能较弱此时可能造成其他服务器的崩溃。
造成上述现象的原因可以归结为以下两点
模块间的耦合性较高例如问题1和2承载能力较弱的模块不具备抗冲击能力。例如问题3 上述的解决方法是在服务器之间加入一个阻塞队列利用生产者和消费者模型解决以上问题。 什么是生产者消费者模型呢如下图
当服务器A接收来自客户端的请求时不把请求直接发给服务器B而是将请求数据加入到队列中服务器B通过队列接收请求并把请求除了的结果返回给A。 当上述分布式系统引入阻塞队列后工作模式如下图所示
引入阻塞队列的好处
解耦合。当服务器A或服务器B出现问题时就不会对其他服务器造成直接的影响当需要添加新的服务器来处理这些请求时新的服务器也同样只需从队列中取数据无需对原有服务器的接口代码进行任何的改动。“削峰填谷”。当服务器A 瞬间接收客户端发来的大量请求时由于服务器B处理请求的速度较慢剩余的请求会在阻塞队列里面堆积虽然客户端获取服务的时间相对增加了但一定程度上缓解了其他承受并发量能力较弱的服务器的压力。
3. Java标准库中阻塞队列的实现 BlockingQueue的主要方法 方法演示如下使用普通入队方法入队4次再使用带有阻塞的出队方法出队4次
public static void main(String[] args) throws InterruptedException {BlockingQueueInteger q new ArrayBlockingQueue(3);System.out.println(数据 5 入队状态: q.offer(5));System.out.println(数据 6 入队状态: q.offer(6));System.out.println(数据 7 入队状态: q.offer(7));System.out.println(数据 8 入队状态: q.offer(8));System.out.print(队列中的数据: );System.out.println(q);System.out.println(数据出队: );for (int i 0; i 4; i) {System.out.print(q.take() );}System.out.println(程序结束 !);
}可以发现当调用 take()方法取出队列元素时因为队列最终为空程序进入了阻塞状态没有打印“程序结束”。 二. 阻塞队列的模拟实现
1. 实现普通队列
阻塞队列的关键方法是两个带有阻塞功能的 put() 和 take()方法而这两个方法是在原有出入队方法上使用 Object类 带有wait()方法 和 notify() 方法让线程进入等待状态 或 唤醒线程。 因此我们可以先把基础的队列进行实现随后在原有基础上进行修改。队列可以使用数组环形队列或链表两种方式实现这里我采用数组的方式实现队列。由于队列的实现方法较为常见这里直接给出实现代码
class MyBlockingQueueE {private Object[] elem;private int defaultCapacity 11; // 阻塞队列默认容量private int front; // 记录队头元素位置private int rear; // 记录队尾元素位置private int size; // 用于记录当前队列元素的实际个数public MyBlockingQueue(){this.elem new Object[defaultCapacity 1];}public MyBlockingQueue(int capacity) {defaultCapacity capacity;this.elem new Object[defaultCapacity 1];}public boolean offer(E val) {// 判断队列是否已满if (size defaultCapacity) {return false;}elem[rear] val;size;// 如果 rear自增 到达数组末尾使 rear 重新到数组的头部rear (rear 1) % (defaultCapacity 1);return true;}public E poll() {// 判断队列是否为空if (front rear) {return null;}Object ret elem[front];size--;// 如果 front 自增 到达数组末尾使 front 重新到数组的头部front (front 1) % (defaultCapacity 1);return (E)ret;}
}2. 实现队列的阻塞功能
当阻塞队列容量已满时调用 put() 方法会进入阻塞状态因此在原先 offer()方法判断的基础上我们需要使用 wait()方法 让线程进入阻塞等待状态考虑到可能有多个线程同时调用 put()方法可能会引起线程安全问题因此我们应在 if()判断条件和整个修改操作上 加锁(或者直接在方法上加锁)。代码如下
public void put (E value) throws InterruptedException {// 判断队列是否已满synchronized (this) {if (size defaultCapacity) {// 队列进入阻塞状态 直到有元素出队时 解除阻塞this.wait();}queue[rear] value;size;// 如果 rear自增 到达数组末尾使 rear 重新到数组的头部rear (rear 1) % (defaultCapacity 1);}
}当队列为空时调用 take() 方法会使线程进入阻塞状态同理若判空条件成立我们需要调用 wait() 方法使线程进入阻塞为防止多个线程在队列即将为空时同时调用 take() 方法引发线程安全问题我们需要在 if()判断语句 和 整个修改操作 进行加锁操作(或者直接在方法上加锁)。代码如下
public E take() throws InterruptedException {// 判断队列是否为空synchronized (this) {if (rear front) {// 队列进入阻塞状态直到有新的元素入队时 解除阻塞this.wait();}Object ret queue[front];// 如果 front 自增 到达数组末尾使 front 重新到数组的头部front (front 1) % (defaultCapacity 1);size--;return (E)ret;}
}3. 解除阻塞状态
什么情况下队列会接触阻塞状态呢
当队满时某个线程从阻塞队列取出一个元素即执行完出队操作后需要使用 notify()方法 唤醒因执行 put()方法而阻塞的线程。当队空时某个线程向队列新增一个元素即执行完入队操作后需要使用 notify()方法唤醒因执行 take()方法而阻塞的线程。
对 put()方法和take()方法 修改后代码如下
public void put (E value) throws InterruptedException {// 判断队列是否已满synchronized (this) {if (size defaultCapacity) {// 队列进入阻塞状态 直到有元素出队时 解除阻塞this.wait();}queue[rear] value;size;// 如果 rear自增 到达数组末尾使 rear 重新到数组的头部rear (rear 1) % (defaultCapacity 1);// 此处的 notify 用来唤醒 队列为空时的 waitthis.notify();}
}public E take() throws InterruptedException {// 判断队列是否为空synchronized (this) {if (rear front) {// 队列进入阻塞状态直到有新的元素入队时 解除阻塞this.wait();}Object ret queue[front];// 如果 front 自增 到达数组末尾使 front 重新到数组的头部front (front 1) % (defaultCapacity 1);size--;// 此处的 notify 用来唤醒 队列为满时的 waitthis.notify();return (E)ret;}
}三. 使用模拟的阻塞队列验证生产者和消费者模型
为了方便看到效果我们假设阻塞队列的容量为2并将生产与消费的数据进行打印。 当生产者与消费者处理数据的频率一样且生产速率为 次/1s、消费速率为 次/1s 时程序的生产与消费数据应轮流打印模拟代码和程序运行结果如下
public static void main(String[] args) {MyBlockingQueueInteger myBlockingQueue new MyBlockingQueue(2);// 生产者Thread producer new Thread(() - {for (int i 0; i 5; i) {try {myBlockingQueue.put(i);System.out.println(生产了: i);Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});// 消费者Thread consumer new Thread(() - {for (int i 0; i 5; i) {try {int ret myBlockingQueue.take();System.out.println(消费了: ret);Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});producer.start();consumer.start();
}当生产速率 消费速率且生产速率为 次/1s、消费速率为 次/2s 时可以预估到当经过5s后程序会因队满进入阻塞状态且后续每消费一次伴随着一次生产为方便观察阻塞情况我们可以在方法实现的地方加上阻塞日志的提示模拟代码和程序运行结果如下
public static void main(String[] args) {MyBlockingQueueInteger myBlockingQueue new MyBlockingQueue(2);Thread producer new Thread(() - {for (int i 0; i 10; i) {try {myBlockingQueue.put(i);System.out.println(生产了: i);Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});Thread consumer new Thread(() - {for (int i 0; i 10; i) {try {int ret myBlockingQueue.take();System.out.println(消费了: ret);Thread.sleep(2000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});producer.start();consumer.start();
}当生产速率 消费速率且生产速率为 次/2s、消费速率为 次/1s 时可以预估到当经过2s后程序会因队满进入阻塞状态且后续每生产一次伴随着一次消费为方便观察阻塞情况我们可以在方法实现的地方加上阻塞日志的提示模拟代码和程序运行结果如下
public static void main(String[] args) {MyBlockingQueueInteger myBlockingQueue new MyBlockingQueue(2);Thread producer new Thread(() - {for (int i 0; i 10; i) {try {myBlockingQueue.put(i);System.out.println(生产了: i);Thread.sleep(2000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});Thread consumer new Thread(() - {for (int i 0; i 10; i) {try {int ret myBlockingQueue.take();System.out.println(消费了: ret);Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});producer.start();consumer.start();
}以上就是本篇文章的全部内容了如果这篇文章对你有些许帮助你的点赞、收藏和评论就是对我最大的支持。 另外文章可能存在许多不足之处也希望你可以给我一点小小的建议我会努力检查并改进。
