前言hello hello这里是洋不写bug~欢迎大家点赞关注收藏铁汁们在上网时可能遇到过服务器因流量突增被瞬间冲垮进而产生宕机例如在一些报名页面查分页面选课页面阻塞队列和消费者生产者模型就是解决这个的核心防护手段这篇博客会从原理解析和代码的实现来解析这部分内容个人主页洋不写bug的博客所属专栏JavaEE学习铁汁们对于JavaEE的各种常用核心语法都可以在上面的前端专栏学习专栏正在持续更新中有问题可以写在评论区或者私信我哦~1阻塞队列我们在数据结构中学习过队列队列具有先进先出的特点队列的底层数据结构就是堆特殊的完全二叉树阻塞队列是一种特殊的队列具有以下特点线程安全普通队列是线程不安全的带有阻塞功能1)如果队列为空尝试出队列就会引发阻塞直到队列不为空2)如果队列满了尝试入队列就会引发阻塞直到队列不满阻塞功能是比较重要的给铁汁们举个例子大家春节都会吃饺子就拿包饺子来举个例子假设有三个滑稽围坐在一张桌子前包饺子滑稽A负责擀饺子皮滑稽B和C负责包饺子这里就可以把滑稽A看作生产者把滑稽B和看作消费者滑稽A擀饺子皮的速度不一定刚好跟滑稽BC包饺子的速度相同这时候搞一个存饺子皮的东西盖帘如下图这个盖帘用于协调工作如果滑稽A包的快了那滑稽A就阻塞等待一会等B和C把盖帘里的饺子皮用的差不多了再开始包如果滑稽A包的慢了那滑稽B和C就阻塞等待一会这些阻塞等待在编程中就要依靠阻塞队列这其实就是生产者消费者模型引入这个模型主要的目的就是为了减少锁竞争因为生产者和消费者的步调不一定总是一致的生产者消费者模型的核心就是阻塞队列阻塞队列可以起到协调的效果2生产者消费者模型生产者消费者模型主要有三个优点可以更好的做到模块之间的解耦合减少资源竞争提升效率削峰填谷在服务器开发上会经常使用这个模型因为一些事情比如网页搜索单靠一台机器是没办法完成的*就需要多个机器合作进行这就需要开发一组服务器程序例如搜索程序的开发如下图用户发出搜索请求调用网关服务器做接入、转发和基础防护不处理核心搜索逻辑网关服务器再调用检索服务器理解用户意图、召回、排序检索服务器再调用用户服务器根据用户的历史记录提供个性化的搜索结果层层调用后再层层返回结果返回给用户这里如果直接调用的话就拿B和C而言B 要调用 C必须知道C 的 IP / 端口、接口地址、请求参数格式、返回结果格式、超时时间等。只要 C 的任何一个细节变了B 的代码必须同步修改否则调用直接失败。加了阻塞队列后C只需要知道怎么从队列中取消息即可就降低了耦合度不加阻塞队列时B 发请求线程死等 C后面处理请求时 线程就不够了就开始在线程池中抢线程加了阻塞队列后B 把任务丢进队列后线程立刻释放去处理下一个搜索请求因此加阻塞队列后能减少资源竞争提升效率最后就是削峰填谷在10年前经常会有选课系统报名系统在使用高峰服务器挂了的情况一个服务器每处理一次请求都是需要消耗一定的硬件资源的cpu内存网络带宽硬盘…一个机器提供的硬件资源数是有限的同一时刻如果请求太多消耗的总资源超出了机器能够提供的资源上限那机器这时候就挂了无法访问了上面在B和C之间加了一个阻塞队列在使用高峰B受到的压力就不会立刻都传给CC会力所能及的处理阻塞队列中的数据不会立刻挂掉那在访问高峰给C加阻塞队列是保护住了那A和B就不会挂掉吗每个服务器完成的功能是不一样的有的服务器处理一个请求消耗的资源多有的消耗的资源少单个资源消耗较多的服务器例如MySQL服务器就比较容易挂掉就需要用阻塞队列来保护但是生产者消费者模型也具有缺点系统更复杂引入的队列层数太多就会增加网络开销3代码实现阻塞队列在Java中就使用BlockingQueueBlockingQueue是一个接口不能直接new要用实现类来newBlockingQueue有两个实现类ArrayBlockingQueue的底层是数组LinkedBlockingQueue的底层是链表后面括号中填上数字表示这个阻塞队列的最大容量代码如下所示publicstaticvoidmain(String[]args){BlockingQueueStringqueuenewArrayBlockingQueue(1000);BlockingQueueStringqueue2newLinkedBlockingQueue(1000);}注出队列和入队列使用的是take和put方法因为有阻塞效果因此是需要抛出异常的虽然也能使用普通队列的offer和add方法因为BlockingQueue接口是继承Queue接口的但是使用offer和add方法是没有阻塞效果的当往队列中加入的元素数量超过队列容量时元素就加不进去了如下所示importjava.util.concurrent.ArrayBlockingQueue;importjava.util.concurrent.BlockingQueue;publicclassDemo28{publicstaticvoidmain(String[]args)throwsInterruptedException{BlockingQueueStringqueuenewArrayBlockingQueue(3);queue.put(aaa);System.out.println(加入一个元素);queue.put(aaa);System.out.println(加入一个元素);queue.put(aaa);System.out.println(加入一个元素);queue.put(aaa);System.out.println(加入一个元素);queue.put(aaa);System.out.println(加入一个元素);}}如果取元素把阻塞队列中的元素给取完了那代码就会卡在这里不执行如下所示importjava.util.concurrent.ArrayBlockingQueue;importjava.util.concurrent.BlockingQueue;publicclassDemo28{publicstaticvoidmain(String[]args)throwsInterruptedException{BlockingQueueStringqueuenewArrayBlockingQueue(3);Stringsqueue.take();System.out.println(取出一个元素s);}}接下来模拟下生产者消费者模型让生产者和消费者线程以相同的速度运行如下所示importjava.util.concurrent.BlockingQueue;importjava.util.concurrent.LinkedBlockingQueue;publicclassDemo29{publicstaticvoidmain(String[]args)throwsInterruptedException{BlockingQueueStringqueuenewLinkedBlockingQueue(1000);ThreadproducernewThread(()-{intcount0;try{while(true){queue.put(count);System.out.println(生产了一个元素count);count;Thread.sleep(1000);}}catch(InterruptedExceptione){e.printStackTrace();}});ThreadconsumernewThread(()-{while(true){try{Stringelemqueue.take();System.out.println(消费了一个元素elem);Thread.sleep(1000);}catch(InterruptedExceptione){e.printStackTrace();}}});producer.start();consumer.start();producer.join();consumer.join();}}如果把生产者的sleep代码删掉那生产者就会以很快的速度来生产消费者还是以正常的速度消费4阻塞队列的模拟实现为了更好的理解阻塞队列的特征我们可以模拟实现一个自己的阻塞队列在数据结构的队列中初始时head和tail都在最前面后来随着元素的添加tail就会向后方移动如果删除元素的话head就会向前移动tail到末尾后需要再重新返回开头head跟tail指向同一个位置时要不就是队列全空要不就是队列全满一会代码我们就可以通过一个size来区分记录当前队列元素的个数进而区分这两种情况不考虑线程安全问题和阻塞基础代码如下所示publicclassMyBlockingQueue{privateString[]arraynull;privateinthead0;privateinttail0;privateintsize0;publicMyBlockingQueue(intcapacity){arraynewString[capacity];}publicvoidput(Stringelem){//这里是防御性编程正常写个等于就可以if(sizearray.length){return;}array[tail]elem;tail;if(tailarray.length){tail0;}size;}publicStringtake(){if(size0){returnnull;}Stringresultarray[head];head;if(headarray.length){head0;}size--;returnresult;}}接下来处理一下线程安全问题对于put操作可能多个线程同时调用put的话put里面的每行代码线程不能插队因此就需要给整个put代码都加上锁这部分不太理解的铁汁可以看一下前面Java线程三中的线程执行顺序问题同理take方法里面的每行代码线程也不能插队因此take方法也要整体加锁那put方法和take方法是必须加同一把锁还是可以加不同的锁这里put和take方法中都有关于修改size的操作这行代码在cpu上不是原子的是分为多条指令的这些指令之间可能会相互插队因此要加同一把锁那线程安全问题解决了怎么来实现阻塞功能呢阻塞队列的规则如下如果队列为空尝试出队列就会触发阻塞直到队列不为空如果队列满了尝试入队列就会触发阻塞直到队列不满这时候就可以巧妙的使用wait和notify在put方法中当阻塞队列满了还有线程往阻塞队列中添加元素时那这个线程就会处于wait状态那什么这个添加元素线程什么时候能正常执行呢当其他线程从阻塞队列中取出元素的时候阻塞队列不满了添加元素的线程就会被激活take方法也是同样的道理当阻塞队列为空时还有线程取元素那这个线程就会处于阻塞状态当下次有线程往阻塞队列中添加元素时这个线程才会被激活也就是put方法中的notify激活调用take方法的线程take方法中的notify激活调用put方法的线程put方法的notify是不会激活调用put方法的wait的因为线程如果在执行put方法时被卡住了就不可能往下执行也就执行不到notify代码如下publicclassMyBlockingQueue{privateString[]arraynull;privateinthead0;privateinttail0;privateintsize0;privateObjectlockernewObject();publicMyBlockingQueue(intcapacity){arraynewString[capacity];}publicvoidput(Stringelem)throwsInterruptedException{synchronized(locker){//这里是防御性编程正常写个等于就可以if(sizearray.length){locker.wait();}array[tail]elem;tail;if(tailarray.length){tail0;}locker.notify();size;}}publicStringtake()throwsInterruptedException{synchronized(locker){if(size0){locker.wait();}Stringresultarray[head];head;if(headarray.length){head0;}size--;locker.notify();returnresult;}}}可以写代码来测试下我们模拟实现的阻塞队列的效果如下所示publicclassDemo30{publicstaticvoidmain(String[]args)throwsInterruptedException{MyBlockingQueuequeuenewMyBlockingQueue(1000);ThreadproducernewThread(()-{intcount0;while(true){try{queue.put(count);System.out.println(生成了一个元素count);count;}catch(InterruptedExceptione){e.printStackTrace();}}});ThreadconsumernewThread(()-{while(true){try{Stringelemqueue.take();System.out.println(消费了一个元素elem);}catch(InterruptedExceptione){e.printStackTrace();}}});producer.start();consumer.start();producer.join();consumer.join();}}另外补充一点在标准库中put方法里面的判断size和array.length的大小写的不是if是while正常进入wait之前当然要进行一次条件判定写成while当唤醒之后还要进行一次条件判定正常来说肯定是条件被打破了才能唤醒此处的条件是size length当其他线程take时size就会小于 length触发notify但是会有一些特殊的代码在唤醒之后仍有size length条件成立的可能性加个while相当于是二次判断更保险了结语阻塞队列和生产者消费者模型在后端开发、高并发业务、中间件、服务开发中都是比较常用的代码并不难理解在日常中只需要知道如何使用即可这里模拟阻塞队列的代码实现只是为了更好的理解以上就是今天的所有内容啦完结撒花