聚合国内IT技术精华文章,分享IT技术精华,帮助IT从业人士成长

并发编程实践之公平有界阻塞队列实现

2021-11-08 14:36 浏览: 4620 次 我要评论(0 条) 字号:



一  背景


JUC 工具包是 JAVA 并发编程的利器。

本文讲述在没有 
JUC 工具包帮助下,借助原生的 JAVA 同步原语, 如何实现一个公平有界的阻塞队列。

希望你也能在文后体会到并发编程的复杂之处,以及 
JUC 工具包的强大。


二  方法


本文使用到的基本工具:

  1. 同步监听器 synchronized ,方法基本和代码块级别;
  2. Object 基础类的 waitnotifynotifyAll;

基于以上基础工具,实现公平有界的阻塞队列,此处:

  1. 将公平的定义限定为 FIFO ,也就是先阻塞等待的请求,先解除等待;
  2. 并不保证解除等待后执行 Action 的先后顺序;
  3. 确保队列的大小始终不超过设定的容量;但阻塞等待的请求数不做限制;


三  实现


1  基础版本


首先,考虑在非并发场景下,借助 ADT 实现一个基础版本

interface Queue {
boolean offer(Object obj);
Object poll();
}class FairnessBoundedBlockingQueue implements Queue { // 当前大小 protected int size;
// 容量 protected final int capacity;
// 头指针,empty: head.next == tail == null protected Node head;
// 尾指针 protected Node tail;
public FairnessBoundedBlockingQueue(int capacity) { this.capacity = capacity; this.head = new Node(null); this.tail = head; this.size = 0; }
// 如果队列已满,通过返回值标识 public boolean offer(Object obj) { if (size < capacity) { Node node = new Node(obj); tail.next = node; tail = node; ++size; return true; } return false; }
// 如果队列为空,head.next == null;返回空元素 public Object poll() { if (head.next != null) { Object result = head.next.value; head.next.value = null; head = head.next; // 丢弃头结点 --size; return result; } return null; }
class Node { Object value; Node next; Node(Object obj) { this.value = obj; next = null; } }}

以上

  1. 定义支持队列的两个基础接口, poll 和 offer;
  2. 队列的实现,采用经典实现;
  3. 考虑在队列空的情况下, poll 返回为空,非阻塞;
  4. 队列在满的情况下, offer 返回 false ,入队不成功,无异常;

需要注意的一点:在出队时,本文通过迁移头结点的方式实现,避免修改尾结点。
在下文实现并发版本时,会看到此处的用意。


2  并发版本


如果在并发场景下,上述的实现面临一些问题,同时未实现给定的一些需求。

通过添加
synchronized ,保证并发条件下的线程安全问题。

注意此处做同步的原因是为了保证类的不变式。


并发问题


在并发场景下,基础版本的实现面临的问题包括:原子性,可见性和指令重排的问题。

参考
JMM 的相关描述。

并发问题,最简单的解决方法是:通过 synchronized 加锁,一次性解决问题。

// 省略接口定义class BoundedBlockingQueue implements Queue {    // 当前大小    protected int size;
// 容量 protected final int capacity;
// 头指针,empty: head.next == tail == null protected Node head;
// 尾指针 protected Node tail;
public BoundedBlockingQueue(int capacity) { this.capacity = capacity; this.head = new Node(null); this.tail = head; this.size = 0; }
// 如果队列已满,通过返回值标识 public synchronized boolean offer(Object obj) { if (size < capacity) { Node node = new Node(obj); tail.next = node; tail = node; ++size; return true; } return false; }
// 如果队列为空,head.next == null;返回空元素 public synchronized Object poll() { if (head.next != null) { Object result = head.next.value; head.next.value = null; head = head.next; // 丢弃头结点 --size; return result; } return null; } // 省略 Node 的定义}

以上,简单粗暴的加 synchronized 可以解决问题,但会引入新的问题:系统活性问题(此问题下文会解决)。

同时,简单加
synchronized 同步是无法实现阻塞等待;即

  1. 如果队列为空,那么出队的动作还是会立即返回,返回为空;
  2. 如果队列已满,那么入队动作还是会立即返回,返回操作不成功;

实现阻塞等待,需要借助 JAVA 中的 PV 原语:wait, notify, notifyAll

参考:JDK 中对 wait, notify, notifyAll 的相关描述。


卫式方法


阻塞等待,可以通过简单的卫式方法来实现,此问题本质上可以抽象为:

  1. 任何一个方法都需要在满足一定条件下才可以执行;
  2. 执行方法前需要首先校验不变式,然后执行变更;
  3. 在执行完成后,校验是否满足后验不变式;

WHEN(condition) Object action(Object arg) {    checkPreCondition();    doAction(arg);    checkPostCondition();}

此种抽象 Ada 在语言层面上实现。在 JAVA 中,借助 wait, notify, notifyAll 可以翻译为:

// 当前线程synchronized Object action(Object arg) {    while(!condition) {        wait();    }    // 前置条件,不变式    checkPreCondition();    doAction();    // 后置条件,不变式    checkPostCondition();}
// 其他线程synchronized Object notifyAction(Object arg) { notifyAll();}

需要注意:

  1. 通常会采用 notifyAll 发送通知,而非 notify ;
    因为如果当前线程收到 notify 通知后被中断,那么系统将一直等待下去。

  2. 如果使用了 notifyAll 那么卫式语句必须放在 while 循环中;
    因为线程唤醒后,执行条件已经不满足,虽然当前线程持有互斥锁。

  3. 卫式条件的所有变量,有任何变更都需要发送 notifyAll 不然面临系统活性问题
据此,不难实现简单的阻塞版本的有界队列,如下

interface Queue {
boolean offer(Object obj) throws InterruptedException;
Object poll() throws InterruptedException;
}class FairnessBoundedBlockingQueue implements Queue { // 当前大小 protected int size;
// 容量 protected final int capacity;
// 头指针,empty: head.next == tail == null protected Node head;
// 尾指针 protected Node tail;
public FairnessBoundedBlockingQueue(int capacity) { this.capacity = capacity; this.head = new Node(null); this.tail = head; this.size = 0; }
// 如果队列已满,通过返回值标识 public synchronized boolean offer(Object obj) throws InterruptedException { while (size < capacity) { wait(); } Node node = new Node(obj); tail.next = node; tail = node; ++size; notifyAll(); // 可以出队 return true; }
// 如果队列为空,阻塞等待 public synchronized Object poll() throws InterruptedException { while (head.next == null) { wait(); } Object result = head.next.value; head.next.value = null; head = head.next; // 丢弃头结点 --size; notifyAll(); // 可以入队 return result; } // 省略 Node 的定义}

以上,实现了阻塞等待,但也引入了更大的性能问题

  1. 入队和出队动作阻塞等待同一把锁,恶性竞争;

  2. 当队列变更时,所有阻塞线程被唤醒,大量的线程上下文切换,竞争同步锁,最终可能只有一个线程能执行;

需要注意的点:

  1. 阻塞等待 wait 会抛出中断异常。关于异常的问题下文会处理;
  2. 接口需要支持抛出中断异常;
  3. 队里变更需要 notifyAll 避免线程中断或异常,丢失消息;


3  锁拆分优化


以上第一个问题,可以通过锁拆分来解决,即:定义两把锁,读锁和写锁;读写分离。

// 省略接口定义class FairnessBoundedBlockingQueue implements Queue {    // 容量    protected final int capacity;
// 头指针,empty: head.next == tail == null protected Node head;
// 尾指针 protected Node tail;
// guard: canPollCount, head protected final Object pollLock = new Object(); protected int canPollCount;
// guard: canOfferCount, tail protected final Object offerLock = new Object(); protected int canOfferCount;
public FairnessBoundedBlockingQueue(int capacity) { this.capacity = capacity; this.canPollCount = 0; this.canOfferCount = capacity; this.head = new Node(null); this.tail = head; }
// 如果队列已满,通过返回值标识 public boolean offer(Object obj) throws InterruptedException { synchronized(offerLock) { while(canOfferCount <= 0) { offerLock.wait(); } Node node = new Node(obj); tail.next = node; tail = node; canOfferCount--; } synchronized(pollLock) { ++canPollCount; pollLock.notifyAll(); } return true; }
// 如果队列为空,阻塞等待 public Object poll() throws InterruptedException { Object result = null; synchronized(pollLock) { while(canPollCount <= 0) { pollLock.wait(); }
result = head.next.value; head.next.value = null; head = head.next; canPollCount--; } synchronized(offerLock) { canOfferCount++; offerLock.notifyAll(); } return result; } // 省略 Node 定义}

以上

  1. 定义了两把锁, pollLock 和 offerLock 拆分出队和入队竞争;

  2. 入队锁同步的变量为:callOfferCount 和 tail;

  3. 出队锁同步的变量为:canPollCount 和 head;

  4. 出队的动作:首先拿到 pollLock 卫式等待后,完成出队动作;
    然后拿到 offerLock 发送通知,解除入队的等待线程。

  5. 入队的动作:首先拿到 offerLock 卫式等待后,完成入队的动作;
    然后拿到 pollLock 发送通知,解除出队的等待线程。

以上实现

  1. 确保通过入队锁和出队锁,分别保证入队和出队的原子性;
  2. 出队动作,通过特别的实现,确保出队只会变更 head ,避免获取 offerLock;
  3. 通过 offerLock.notifyAll 和 pollLock.notifyAll 解决读写竞争的问题;

但上述实现还有未解决的问题:

当有多个入队线程等待时,一次出队的动作会触发所有入队线程竞争,大量的线程上下文切换,最终只有一个线程能执行。

即,还有 读与读 和 写与写 之间的竞争问题。


4  状态追踪解除竞争


此处可以通过状态追踪,解除读与读之间和写与写之间的竞争问题

class FairnessBoundedBlockingQueue implements Queue {    // 容量    protected final int capacity;
// 头指针,empty: head.next == tail == null protected Node head;
// 尾指针 protected Node tail;
// guard: canPollCount, head protected final Object pollLock = new Object(); protected int canPollCount; protected int waitPollCount;
// guard: canOfferCount, tail protected final Object offerLock = new Object(); protected int canOfferCount; protected int waitOfferCount;
public FairnessBoundedBlockingQueue(int capacity) { this.capacity = capacity; this.canPollCount = 0; this.canOfferCount = capacity; this.waitPollCount = 0; this.waitOfferCount = 0; this.head = new Node(null); this.tail = head; }
// 如果队列已满,通过返回值标识 public boolean offer(Object obj) throws InterruptedException { synchronized(offerLock) { while(canOfferCount <= 0) { waitOfferCount++; offerLock.wait(); waitOfferCount--; } Node node = new Node(obj); tail.next = node; tail = node; canOfferCount--; } synchronized(pollLock) { ++canPollCount; if (waitPollCount > 0) { pollLock.notify(); } } return true; }
// 如果队列为空,阻塞等待 public Object poll() throws InterruptedException { Object result; synchronized(pollLock) { while(canPollCount <= 0) { waitPollCount++; pollLock.wait(); waitPollCount--; }
result = head.next.value; head.next.value = null; head = head.next; canPollCount--; } synchronized(offerLock) { canOfferCount++; if (waitOfferCount > 0) { offerLock.notify(); } } return result; } // 省略 Node 的定义}

以上

  1. 通过 waitOfferCount 和 waitPollCount 的状态追踪解决 读写内部的竞争问题;
  2. 当队列变更时,根据追踪的状态,决定是否派发消息,触发线程阻塞状态解除;

但,上述的实现在某些场景下会运行失败,面临活性问题,考虑

情况一:

  1. 初始状态队列为空 线程 A 执行出队动作,被阻塞在 pollLock , 此时 waitPollCount==1;

  2. 此时线程 A 在执行 wait 时被中断,抛出异常, waitPollCount==1 并未被重置;

  3. 阻塞队列为空,但 waitPollCount==1 类状态异常;

情况二:

  1. 初始状态队列为空 线程 A B 执行出队动作,被阻塞在 pollLock , 此时 waitPollCount==2;

  2. 线程 C 执行入队动作,可以立即执行,执行完成后,触发 pollLock 解除一个线程等待 notify;

  3. 触发的线程在 JVM 实现中是随机的,假设线程 A 被解除阻塞;

  4. 假设线程 A 在阻塞过程中已被中断,阻塞解除后 JVM 检查 interrupted 状态,抛出 InterruptedException 异常;

  5. 此时队列中有一个元素,但线程 A 仍阻塞在 pollLock 中,且一直阻塞下去;

以上为解除阻塞消息丢失的例子,问题的根源在与异常处理。


5  解决异常问题


解决线程中断退出的问题,线程校验中断状态的场景

  1. JVM 通常只会在有限的几个场景检测线程的中断状态, wait, Thread.join, Thread.sleep;


  2. JVM 在检测到线程中断状态 Thread.interrupted() 后,会清除中断标志,抛出 InterruptedException;


  3. 通常为了保证线程对中断及时响应, run 方法中需要自主检测中断标志,中断线程,特别是对中断比较敏感需要保持类的不变式的场景;

class FairnessBoundedBlockingQueue implements Queue {    // 容量    protected final int capacity;
// 头指针,empty: head.next == tail == null protected Node head;
// 尾指针 protected Node tail;
// guard: canPollCount, head, waitPollCount protected final Object pollLock = new Object(); protected int canPollCount; protected int waitPollCount;
// guard: canOfferCount, tail, waitOfferCount protected final Object offerLock = new Object(); protected int canOfferCount; protected int waitOfferCount;
public FairnessBoundedBlockingQueue(int capacity) { this.capacity = capacity; this.canPollCount = 0; this.canOfferCount = capacity; this.waitPollCount = 0; this.waitOfferCount = 0; this.head = new Node(null); this.tail = head; }
// 如果队列已满,通过返回值标识 public boolean offer(Object obj) throws InterruptedException { if (Thread.interrupted()) { throw new InterruptedException(); // 线程已中断,直接退出即可,防止中断线程竞争锁 } synchronized(offerLock) { while(canOfferCount <= 0) { waitOfferCount++; try { offerLock.wait(); } catch (InterruptedException e) { // 触发其他线程 offerLock.notify(); throw e;
} finally { waitOfferCount--; } } Node node = new Node(obj); tail.next = node; tail = node; canOfferCount--; } synchronized(pollLock) { ++canPollCount; if (waitPollCount > 0) { pollLock.notify(); } } return true; }
// 如果队列为空,阻塞等待 public Object poll() throws InterruptedException { if (Thread.interrupted()) { throw new InterruptedException(); } Object result = null; synchronized(pollLock) { while(canPollCount <= 0) { waitPollCount++; try { pollLock.wait(); } catch (InterruptedException e) { pollLock.notify(); throw e; } finally { waitPollCount--; } }
result = head.next.value; head.next.value = 0; // ignore head; head = head.next; canPollCount--; } synchronized(offerLock) { canOfferCount++; if (waitOfferCount > 0) { offerLock.notify(); } } return result; } // 省略 Node 的定义}

以上

  1. 当等待线程中断退出时,捕获中断异常,通过 pollLock.notify 和 offerLock.notify 转发消息;

  2. 通过在 finally 中恢复状态追踪变量;

通过状态变量追踪可以解决读与读之间和写与写之间的锁竞争问题。

以下考虑如果解决读与读之间和写与写之间的公平性问题。


6  解决公平性


公平性的问题的解决需要将状态变量的追踪转换为:请求监视器追踪。

  1. 每个请求对应一个监视器;
  2. 通过内部维护一个 FIFO 队列,实现公平性;
  3. 在队列状态变更时,释放队列中的监视器;

以上逻辑可以统一抽象为

boolean needToWait;synchronized(this) {    needToWait = calculateNeedToWait();    if (needToWait) {        enqueue(monitor); // 请求对应的monitor    }}if (needToWait) {    monitor.doWait();}

需要注意

  1. monitor.doWait() 需要在 this 的卫式语句之外,因为如果在内部, monitor.doWait 并不会释放 this锁;


  2. calculateNeedToWait() 需要在 this 的守卫之内完成,避免同步问题;


  3. 需要考虑中断异常的问题;

基于以上的逻辑抽象,实现公平队列

// 省略接口定义class FairnessBoundedBlockingQueue implements Queue {    // 容量    protected final int capacity;
// 头指针,empty: head.next == tail == null protected Node head;
// 尾指针 protected Node tail;
// guard: canPollCount, head, pollQueue protected final Object pollLock = new Object(); protected int canPollCount;
// guard: canOfferCount, tail, offerQueue protected final Object offerLock = new Object(); protected int canOfferCount;
protected final WaitQueue pollQueue = new WaitQueue(); protected final WaitQueue offerQueue = new WaitQueue();
public FairnessBoundedBlockingQueue(int capacity) { this.capacity = capacity; this.canOfferCount = capacity; this.canPollCount = 0; this.head = new Node(null); this.tail = head; }
// 如果队列已满,通过返回值标识 public boolean offer(Object obj) throws InterruptedException { if (Thread.interrupted()) { throw new InterruptedException(); // 线程已中断,直接退出即可,防止中断线程竞争锁 } WaitNode wait = null; synchronized(offerLock) { // 在有阻塞请求或者队列为空时,阻塞等待 if (canOfferCount <= 0 || !offerQueue.isEmpty()) { wait = new WaitNode(); offerQueue.enq(wait); } else { // continue. } }
try { if (wait != null) { wait.doWait(); } if (Thread.interrupted()) { throw new InterruptedException(); } } catch (InterruptedException e) { offerQueue.doNotify(); throw e; }
// 确保此时线程状态正常,以下不会校验中断 synchronized(offerLock) { Node node = new Node(obj); tail.next = node; tail = node; canOfferCount--; } synchronized(pollLock) { ++canPollCount; pollQueue.doNotify(); } return true; }
// 如果队列为空,阻塞等待 public Object poll() throws InterruptedException { if (Thread.interrupted()) { throw new InterruptedException(); } Object result = null; WaitNode wait = null; synchronized(pollLock) { // 在有阻塞请求或者队列为空时,阻塞等待 if (canPollCount <= 0 || !pollQueue.isEmpty()) { wait = new WaitNode(); pollQueue.enq(wait); } else { // ignore } }
try { if (wait != null) { wait.doWait(); } if (Thread.interrupted()) { throw new InterruptedException(); } } catch (InterruptedException e) { // 传递消息 pollQueue.doNotify(); throw e; }
// 以下不会检测线程中断状态 synchronized(pollLock) { result = head.next.value; head.next.value = 0; // ignore head; head = head.next; canPollCount--; }
synchronized(offerLock) { canOfferCount++; offerQueue.doNotify(); } return result; }
class WaitQueue {
WaitNode head; WaitNode tail;
WaitQueue() { head = new WaitNode(); tail = head; }
synchronized void doNotify() { for(;;) { WaitNode node = deq(); if (node == null) { break; } else if (node.doNotify()) { // 此处确保NOTIFY成功 break; } else { // ignore, and retry. } } }
synchronized boolean isEmpty() { return head.next == null; }
synchronized void enq(WaitNode node) { tail.next = node; tail = tail.next; }
synchronized WaitNode deq() { if (head.next == null) { return null; } WaitNode res = head.next; head = head.next; if (head.next == null) { tail = head; // 为空,迁移tail节点 } return res; } }
class WaitNode { boolean released; WaitNode next; WaitNode() { released = false; next = null; }
synchronized void doWait() throws InterruptedException { try { while (!released) { wait(); } } catch (InterruptedException e) { if (!released) { released = true; throw e; } else { // 如果是NOTIFY之后收到中断的信号,不能抛出异常;需要做RELAY处理 Thread.currentThread().interrupt(); } } }
synchronized boolean doNotify() { if (!released) { released = true; notify(); // 明确释放了一个线程,返回true return true; } else { // 没有释放新的线程,返回false return false; } } } // 省略 Node 的定义}

以上

  1. 核心是替换状态追踪变量为同步节点, WaitNode


  2. WaitNode 通过简单的同步队列组织实现 FIFO 协议,每个线程等待各自的 WaitNode 监视器;

  3. WaitNode 内部维持 released 状态,标识线程阻塞状态是否被释放,主要是为了处理中断的问题;

  4. WaitQueue 本身是全同步的,由于已解决了读写竞争已经读写内部竞争的问题, WaitQueue 同步并不会造成问题;

  5. WaitQueue 是无界队列,是一个潜在的问题;但由于其只做同步的追踪,而且追踪的通常是线程,通常并不是问题;

  6. 最终的公平有界队列实现,无论是入队还是出队,首先卫式语句判定是否需要入队等待,如果入队等待,通过公平性协议等待;
    当信号释放时,借助读写锁同步更新队列;最后同样借助读写锁,触发队列更新消息;


7  等待时间的问题


并发场景下,等待通常会设置为限时等待 TIMED_WAITING ,避免死锁或损失系统活性;

实现同步队列的限时等待,并没想象的那么困难

class TimeoutException extends InterruptedException {}
class WaitNode { boolean released; WaitNode next; WaitNode() { released = false; next = null; }
synchronized void doWait(long milliSeconds) throws InterruptedException { try { long startTime = System.currentTimeMillis(); long toWait = milliSeconds; for (;;) { wait(toWait); if (released) { return; } long now = System.currentTimeMillis(); toWait = toWait - (now - startTime); if (toWait <= 0) { throw new TimeoutException(); } } } catch (InterruptedException e) { if (!released) { released = true; throw e; } else { // 如果已经释放信号量,此处不抛出异常;但恢复中断状态 Thread.currentThread().interrupt(); } } }
synchronized boolean doNotify() { if (!released) { released = true; notify(); return true; } else { return false; } }

由于所有的等待都阻塞在 WaitNode 监视器,以上

  • 首先定义超时异常,此处只是为了方便异常处理,继承 InterruptedException


  • 此处依赖于 wait(long timeout) 的超时等待实现,这通常不是问题;


最后,将 WaitNode 超时等待的逻辑,带入到 FairnessBoundedBlockingQueue 实现中,即可。


四  总结


本文通过一步步迭代,最终借助 JAVA 同步原语实现初版的公平有界队列。迭代实现过程中可以看到以下几点:

  1. 观念的转变,将调用一个类的方法思维转换为:在满足一定条件下方法才可以调用,在调用前需要满足不变式,调用后满足不变式;
    由于并发的问题很难测试,通常要采用卫式表达证明并发的正确性;


  2. 在迭代实现中会看到很多模式,比如,读写分离时,其实可以抽象为读锁和写锁;就得到了一个抽象的 Lock 的定义;
    比如,读写状态追踪,可以采用 Exchanger 抽象表达;


  3. 另外,本文的实现远非完善,还需要考虑支持 Iterator 遍历、状态查询及数据迁移等操作;

最后,相信大家再看 JUC 的工具包实现,定有不一样的体会。



阿里云资源编排ROS使用教程


资源编排(Resource Orchestration)是一种简单易用的云计算资源管理和自动化运维服务。用户通过模板描述多个云计算资源的依赖关系、配置等,并自动完成所有资源的创建和配置,以达到自动化部署、运维等目的。编排模板同时也是一种标准化的资源和应用交付方式,并且可以随时编辑修改,使基础设施即代码(Infrastructure as Code)成为可能。

点击阅读原文查看详情!



网友评论已有0条评论, 我也要评论

发表评论

*

* (保密)

Ctrl+Enter 快捷回复