阻塞的本质——等待队列

Posted 2023-05-09

参考技术A

写程序的时候，我们常常说某个系统调用是阻塞调用。从用户层的角度，基本理解是：进程在执行某个系统调用的时候，因为需要的资源不满足（IO操作，加锁等等），导致进程“停”在那里。等到资源就绪了或者设置的timeout时间超时了，进程得以继续执行。

从内核的角度，面对用户层对阻塞调用的需求，需要实现哪些机制呢？

1.首先，进程陷入内核，内核发现进程所要求的资源暂时无法满足，需要将其设置为睡眠状态，然后调度其他进程执行。这里引出一个问题：（1）内核如何将一个进程睡眠？
2.再来，等待资源就绪时，我们需要唤醒等待在该资源上面的进程。这里存在两个问题：（2）内核是怎么知道资源就绪的？以及，（3）某个资源就绪了，内核怎么找到对应等待的进程的？
问题一：内核如何将一个进程睡眠

进程的task_struct结构有一个状态成员。将其设置为“睡眠”，并将task_struct结构从就绪队列中移走，内核就不会调度其执行，也就相当于睡眠。

问题二：内核怎么知道资源就绪的

中断。 内核的所有的工作都是由中断驱动的。 不管是系统调用陷入内核，还是调度，还是其他的内核活动，都是由各种各样的中断来触发执行的。对于设备IO，如果设备空闲了，会触发一个外部中断，该中断触发内核执行处理程序，通知等待进程、执行回调等等。

问题三：资源就绪了，内核怎么找到对应等待的进程

答案是等待队列。我们将一个资源和一个等待队列关联起来。如果进程所请求的资源还未就绪，就先加入到该资源的等待队列中。等到资源就绪了，就唤醒等待队列中的进程，加入到调度。

等待队列就是一个普通的双向链表，该链表的每个节点都代表一个进程task_struct的封装。每个资源都会有相应的等待队列。

“惊群”的基本行为是：有多个进程或者线程等待在同一个资源上，而且该资源一次只能有一个进程处理，比如文件描述符的写操作，accept一个新连接等。那么在资源就绪的时候。如果内核采取的策略是唤醒所有的进程。这样，只有一个进程获取了该资源，其他进程发现没有资源就绪，继续进入睡眠（所谓虚假唤醒）。这样的行为浪费了系统的CPU资源。

那是不是，内核在资源就绪的时候，就唤醒一个进程不就得了。其实也不是，因为不是所有资源都是互斥的。比如，某个文件的读操作。

那么，惊群问题怎么解决？

在用户态，可以有不同的解决方式。或者忽略惊群所带来的开销，或者使用锁方式保证一次只有一个进程来阻塞在一个资源上。而对于内核来说，在等待队列上增加了一个是否“互斥等待”的标志。

如果是互斥等待的，一次唤醒一个进程。如果不是互斥等待的，一次唤醒所有进程。

互斥等待的经典例子：accept。因为我们很明确知道，对一个listen fd的accept,肯定是一次只有一个进程可以处理。那么，我们在listen fd上的等待队列，就毫无疑问可以设置为“互斥等待”。所以，现今版本的linux内核，解决了accept的惊群问题。

但是像epoll_wait的惊群问题，就无法从等待队列的互斥等待来解决。首先，epoll fd上也有一个等待队列，代表epoll fd所监听的其他若干文件描述符（资源）就绪时，唤醒等待队列上的资源。因为我们无法确定，这些资源是不是都是互斥访问的，还是都不是。所以，只好唤醒所有进程。更多的惊群问题，可以查阅相关资料。

https://shunlqing.github.io/2018/05/19/2018_5_19LinuxKernel_WaitQueue/

Java多线程：阻塞队列与等待唤醒机制初探

文章目录

• 阻塞队列概述
• 阻塞队列基本使用
• • BlockingQueue
  • • BlockingQueue的核心方法:
    • 代码示例
• 阻塞队列实现等待唤醒机制
• • 代码需求：
  • • 代码示例

阻塞队列概述

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作是：在队列为空时，获取元素的线程会等待队列变为非空。当队列满时，存储元素的线程会等待队列可用。阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。

阻塞队列基本使用

BlockingQueue

常见BlockingQueue有:

• ArrayBlockingQueue: 底层是数组,有界
• LinkedBlockingQueue: 底层是链表,无界.但不是真正的无界,最大为int的最大值

他们两个和List的实现子类有点像。

他们的继承结构如下：

BlockingQueue的核心方法:

核心方法：

方法	说明
put(anObject)	将参数放入队列,如果放不进去会阻塞
take()	取出第一个数据,取不到会阻塞

其他方法（与List类似）

方法\\处理方式	抛出异常	返回特殊值	一直阻塞	超时退出
插入方法	add(e)	offer(e)	put(e)	offer(e,time,unit)
移除方法	remove()	poll()	take()	poll(time,unit)
检查方法	element()	peek()	不可用	不可用

多线程一般多用put与take

代码示例

class Demo02 
    public static void main(String[] args) throws Exception 
        // 创建阻塞队列的对象,容量为 1
        ArrayBlockingQueue<String> arrayBlockingQueue = new ArrayBlockingQueue<>(3);

        // 存储元素
        arrayBlockingQueue.put("糖果");

        // 取元素
        System.out.println(arrayBlockingQueue.take());
        System.out.println(arrayBlockingQueue.take()); // 取不到会阻塞

        System.out.println("程序结束了");
    

程序没有结束，他会一直读取阻塞队列

阻塞队列实现等待唤醒机制

代码需求：

生产者类(Cooker)：实现Runnable接口，重写run()方法，设置线程任务

1. 构造方法中接收一个阻塞队列对象
2. 在run方法中循环向阻塞队列中添加包子
3. 打印添加结果

消费者类(Foodie)：实现Runnable接口，重写run()方法，设置线程任务

1. 构造方法中接收一个阻塞队列对象
2. 在run方法中循环获取阻塞队列中的包子
3. 打印获取结果

测试类(Demo)：里面有main方法，main方法中的代码步骤如下

1. 创建阻塞队列对象
2. 创建生产者线程和消费者线程对象,构造方法中传入阻塞队列对象
3. 分别开启两个线程

代码示例

package com.test;


import java.util.concurrent.ArrayBlockingQueue;

class Cooker extends Thread 

    private ArrayBlockingQueue<String> bd;

    public Cooker(ArrayBlockingQueue<String> bd) 
        this.bd = bd;
    

    @Override
    public void run() 
        while (true) 
            try 
                bd.put("糖果");
                System.out.println("厨师放入一个糖果");
             catch (InterruptedException e) 
                e.printStackTrace();
            
        
    


class Foodie extends Thread 
    private ArrayBlockingQueue<String> bd;

    public Foodie(ArrayBlockingQueue<String> bd) 
        this.bd = bd;
    

    @Override
    public void run() 
        while (true) 
            try 
                String take = bd.take();
                System.out.println("吃货将" + take + "拿出来吃了");
             catch (InterruptedException e) 
                e.printStackTrace();
            
        

    


class Demo 
    public static void main(String[] args) 
        // 一个阻塞队列代表一个桌子
        ArrayBlockingQueue<String> bd = new ArrayBlockingQueue<>(1);

        Foodie f = new Foodie(bd);
        Cooker c = new Cooker(bd);

        f.start();
        c.start();