Concurrency(十一: 饥饿与公平)

如果一个线程因为其他线程占满了而无法获取CPU运行时间,这种情况我们称之为“饥饿现象”.线程将一直饥饿下去,因为其他线程总能替代它获取CPU运行时间.解决这种情况的措施我们称之为“公平措施”.即让所有线程都能获得一次执行的机会.

Java中导致饥饿现象的原因

在Java中,以下三种情况能够产生饥饿现象:

1. 优先级高的线程蚕食了优先级低的线程的所有CPU运行时间.
2. 线程无限期的阻塞进入同步代码块因为其他线程总是能够在它之前进入.
3. 线程无限期的等待(调用wait())唤醒因为其他线程总是能够在它之间被唤醒(接收到notify()信号).

优先级高的线程蚕食了优先级低的线程的所有CPU运行时间

你可以分别对每一个线程设置优秀级.高优先级的线程能够获取到更多的CPU运行时间.你可以为线程设置1~10的优先级,但这完全依赖于应用运行在哪个操作系统之上.对于大多数应用采用默认优秀级就好.

线程无限期的阻塞进入同步代码块

Java的同步代码块是引起饥饿现象的另一个原因.Java同步代码块没办法保证在等待中的线程能够按照某种序列进入同步代码块.这意味着理论上会有一个线程无限期的等待进入同步代码块,因为其他线程总是能够在它之前进入同步代码块.这种问题也称之为"饥饿现象",该线程将会一直饥饿下去,因为其他线程总能替代它获取CPU运行时间.

线程无限期的等待唤醒

在多个线程同时调用同一个对象的 wait() 方法时, notify() 方法无法保证唤醒哪个线程.这会导致某些线程一直在等待.这会产生一个线程一直在等待唤醒的风险,因为其他线程总能在它之前被唤醒.

Java实现公平措施

虽然在Java中不可能实现百分百的公平措施,但我们仍然可以实现自己的同步器结构来增加线程间的公平性.

我们先来学习一个简单的同步器代码块:

public class Synchronizer{
  public synchronized void doSynchronized(){
    // 花费相当长的时间来执行工作
  }
}
复制代码

如果有多于一个的线程调用doSynchronized方法,那么其他线程都需要等待第一个进入同步代码块的线程退出方法.只要有多于一个线程在等待状态,就不能保证哪个线程先被允许进入同步代码块.

使用锁来代替同步代码块

为了增加等待线程的公平性,首先我们需要使用锁来代替同步代码块来实现同步机制.

public class Synchronizer{
    Lock lock = new Lock();
    
    public void doSynchronized() throws InterruptedException{
        this.lock.lock();
        // 临界区代码, 花费相当长的时间来执行工作
        this.lock.unLock();
    }
}
复制代码

我们注意到doSynchronized不再使用 synchronized 来声明.取而代之的是将临界区代码放置在lock.lock()和lock.unLock()方法调用之间.

一个简单Lock类实现:

public class Lock{
public class Lock {
    private boolean isLocked = false;
    private Thread lockingThread;

    public synchronized void lock() throws InterruptedException {
        while (isLocked){
            wait();
        }
        isLocked = true;
        lockingThread = Thread.currentThread();
    }

    public synchronized void unLock(){
        if(lockingThread != null && lockingThread == Thread.currentThread()){
            throw new IllegalMonitorStateException("Calling thread is not locked this lock");
        }
        isLocked = false;
        lockingThread = null;
        notify();
    }
}
复制代码

如果你看了上文的Synchronizer和现在的Lock实现,你会发现如果有多于一个线程同时访问lock()方法时,会发生阻塞.其次,如果当前的锁为锁住状态的话,线程会在lock方法中的while循环内部调用wait()方法从而进入等待状态.记住,一个线程一旦调用wait()完毕,则会释放当前对象锁,让其他线程可以进入lock()方法.最后的结果是多个线程进入lock()方法的while循环中调用wait()方法进入等待状态.

如果你回头看doSynchronized()方法,你会发现lock()和unlock()状态切换中间的注释,这将花费相当长的时间来执行两个方法调用间的代码.需要我们确认的是执行这些代码需要花费相当长的时间来比较进入lock()方法和在锁被锁住的情况下调用wait()方法.这意味着大部分时间都用来等待锁的锁住和在lock()方法内部调用的wait()方法完毕后等待退出wait()方法.而不是等待进入lock()方法.

在之前的同步代码块的状态下,如果有多个线程等待进入同步代码块,无法保证哪个线程先进入同步代码块.同样的在调用wait()方法进入等待状态后,无法保证调用notify()后哪个线程会先被唤醒.所以当前Lock的实现版本并不比之前的synchronized版本的doSynchronized()公平到哪去.但我们可以稍作更改.

当前Lock版本调用的是它自己的wait()方法.如果将每个线程调用的wait()方法替换成不同对象的.即每个线程对应调用一个对象的wait()方法,即Lock能够决定在往后的时间里到底调用哪个对象的notify()方法,这样就能具体有效的选择唤醒哪个线程.

一个公平锁的实现

以下内容是将上文提及的Lock.class转变为一个公平锁即FairLock.class.你会发现与之前版本对比,这个实现仅是调整了同步代码块和wait()/notify()的调用方式而已.

实际上在得到当前这个版本的公平锁之前遇到了许许多多的问题,而每解决这其中的一个问题都需要长篇概论来阐述,解决这些问题的每一个步骤都会在往后的主题中提及.这包含嵌套监控器锁死, 滑动条件和信号丢失问题. 现在重点要知道的是线程以队列的方式来调用lock()方法中的wait()方法,且每次在公平锁未锁住时仅能让队列头部的线程来获取和锁住公平锁实例.其他线程则处在等待状态直到进入队列头部.

public class FairLock {
    private boolean isLocked = false;
    private Thread lockingThread;
    private List<QueueObject> waitingThreads = new ArrayList<>();

    public void lock() throws InterruptedException {
        // 1. 为每个线程创建一个QueueObject
        QueueObject queueObject = new QueueObject();
        boolean isLockedForThisThread = true;
        synchronized (this) {
            // 2. 添加当前线程的QueueObject到队列中
            waitingThreads.add(queueObject);
        }

        while (isLockedForThisThread) {
            synchronized (this) {
                isLockedForThisThread = isLocked || waitingThreads.get(0) != queueObject;
                if (!isLockedForThisThread) {
                    // 3. 锁住当前公平锁
                    isLocked = true;
                    waitingThreads.remove(queueObject);
                    lockingThread = Thread.currentThread();
                    return;
                }
            }
            try {
                // 4. 调用该线程对应QueueObject的wait()方法进入等待状态
                queueObject.doWait();
            } catch (InterruptedException e) {
                synchronized (this) {
                    waitingThreads.remove(queueObject);
                }
                throw e;
            }
        }
    }

    public synchronized void unlock() {
        if (this.lockingThread != Thread.currentThread()) {
            throw new IllegalMonitorStateException("Calling thread has not locked this lock");
        }
        // 1. 释放公平锁
        isLocked = false;
        lockingThread = null;
        if (waitingThreads.size() > 0) {
            // 2. 调用队列头部线程一一对应的QueueObject唤醒线程
            waitingThreads.get(0).doNotify();
        }
    }
}
复制代码

public class QueueObject {
    private boolean isNotified = false;

    public synchronized void doWait() throws InterruptedException {
        while (!isNotified) {
            this.wait();
        }
        this.isNotified = false;
    }

    public synchronized void doNotify() {
        this.isNotified = true;
        this.notify();
    }

    public boolean equals(Object o) {
        return this == o;
    }
}
复制代码

首先你会注意到lock不再声明为 synchronized .取而代之的是将需要做同步限制的代码块嵌套到 synchronized 代码块中.

每个线程调用lock()方法后都会创建与之对应的 QueueObject 实例,并进入到队列中.线程调用unlock()方法后会从队列的头部取得 QueueObject 对象并调用它的doNotify()方法来唤醒与之对应的线程.对于所有等待的线程来说,这种方式每次仅会唤醒一个线程.这部分就是FairLock用来确保公平的代码.

我们注意到lock的锁住状态会在同一个代码块中不停的检查和设置来解决滑动条件带来的问题.

同时我们注意到 QueueObject 就是一个 Semaphore .doWait()和doNotify()调用所产生的状态会存储在 QueueObject 内部.这用来解决信号丢失问题,即一个线程在调用queueObject().doWait()时,被另一个线程抢先机会调用了unlock()中的queueObject.doNotify(). queueObject.doWait()调用被放置在 synchronized(this) 同步代码块之外,用于解决嵌套监控器锁死问题.这样当没有线程在lock()方法的 synchronized(this) 代码块中执行时,其他线程可以正常调用unLock()方法.

最后需要注意的是,为什么需要将queueObject.doWait()调用放置在try-catch中.当线程通过抛出 InterruptedException 来终止lock()方法调用时,我们需要将线程与之对应的 QueueObject 踢出队列.