我丢!ReentrantLock 公平锁竟然不公平了?

  |   0 评论   |   0 浏览

大家好,我是一航!

前些天,有小伙伴在群里面发了这样一个问题,意思是:Java的公平锁为什么只有在 Linux 系统上才是公平的,Windows 却不公平?,并贴出了他测试的源码。

当看到这个问题,第一时间浮现在脑海的就是三个字:不可能,Java 的公平锁肯定是公平的,那这位同学为啥会说 Windows 下不公平呢?把他给的源码跑了跑:

public class Demo06 {
    public static void main(String[] args) {
        Demo06Service service = new Demo06Service(true);        // 公平锁
//        Demo06Service service = new Demo06Service(false);        // 非公平锁

        Thread[] threads = new Thread[30];
        for (int i = 0; i < threads.length; i++) {
            threads[i] = new Demo06Thread(service);
        }
        for (int i = 0; i < threads.length; i++) {
            threads[i].start();
        }
    }
}

class Demo06Service{
    private Lock lock;
    public Demo06Service(boolean isFair){
        // 无参的ReentrantLock是非公平锁,通过boolean参数控制锁的类型,如果是true会使用公平锁,否则就是非公平锁
        lock = new ReentrantLock(isFair);
    }
    public void foo(){
        try{
            lock.lock();
            System.out.println(Thread.currentThread().getName() + "获得锁定");
        }finally {

            lock.unlock();
        }

    }
}

class Demo06Thread extends Thread{
    private Demo06Service service;
    public Demo06Thread(Demo06Service service){
        this.service = service;
    }

    @Override
    public void run() {
        service.foo();
    }
}

代码逻辑也不复杂,预先创建了30个线程(Demo06Thread),循环启动(start)线程去执行 Demo06Servicefoo方法,方法的代码上加了 ReentrantLock公平锁;

Windows 和 Linux 测试效果如下:

而这位同学不理解的是:**既然我是按着顺序启动(start)的线程,方法的执行也加上了公平锁,结果为什么只有 Linux 是按着0-29的顺序在输出呢?**因此让这位同学开始怀疑,难道是 ReentrantLock 公平锁在 Windows 机器上变的不公平了?

很明显,这位同学将多线程start的顺序当成了最终执行的顺序,而线程的执行顺序和start的顺序是没有直接关系的

但问题是,似乎在 Linux 机器上面测试的时候,已经达到了预期,只有在 Windows 才错乱,这不就是 Windows 不公平嘛!

注意,这里只是个假象,在 Linux 机器,线程数量变多之后,顺序也会发生变化,同样也会乱序

因此,这个现象跟 ReentrantLock 的公平锁没有任何关系,ReentrantLock 公平锁肯定是公平的,问题主要是出在线程执行以及操作系统CPU竞争策略上;

下面就一起来分析一下具体原因;

Thread.start() 和 Thread.run()

用过线程的同学,就肯定知道这两个方法,作用就是执行线程的 run 方法中的代码,但是本质上有很大的区别

Thread.start()

源码如下:

/**
 * Causes this thread to begin execution; the Java Virtual Machine
 * calls the <code>run</code> method of this thread.
 * <p>
 * The result is that two threads are running concurrently: the
 * current thread (which returns from the call to the
 * <code>start</code> method) and the other thread (which executes its
 * <code>run</code> method).
 * <p>
 * It is never legal to start a thread more than once.
 * In particular, a thread may not be restarted once it has completed
 * execution.
 *
 * @exception  IllegalThreadStateException  if the thread was already
 *               started.
 * @see        #run()
 * @see        #stop()
 */
public synchronized void start() {
    /**
     * This method is not invoked for the main method thread or "system"
     * group threads created/set up by the VM. Any new functionality added
     * to this method in the future may have to also be added to the VM.
     *
     * A zero status value corresponds to state "NEW".
     */
    if (threadStatus != 0)
        throw new IllegalThreadStateException();

    /* Notify the group that this thread is about to be started
     * so that it can be added to the group's list of threads
     * and the group's unstarted count can be decremented. */
    group.add(this);

    boolean started = false;
    try {
        start0();
        started = true;
    } finally {
        try {
            if (!started) {
                group.threadStartFailed(this);
            }
        } catch (Throwable ignore) {
            /* do nothing. If start0 threw a Throwable then
              it will be passed up the call stack */
        }
    }
}

private native void start0();

start() 是一个 synchronized修饰的方法,最终是调用了一个 start0()native方法,此时会执行以下步骤:

  • 通过JVM告诉操作系统创建一个Thread;
  • 操作系统开辟内存,并通过创建 Thread 线程对象;
  • 操作系统对 Thread 对象进行调度,并确定执行时间;
  • Thread 在操作系统中被执行;

以上过程,主线程(main)并不需要等待 Thread 执行完毕,而是直接往后继续执行了;至于对 Thread 的调度,什么时候执行,就完全交由系统根据自己的CPU竞争策略来决定,跟主线程(main)、调用 start 方法的顺序没有任何关系了;

CPU 竞争策略

那上面的测试输出,为什么最终执行的结果看起来只有Windows乱了呢?

这是由于两个操作系统采用了不同的 CPU 竞争策略,Linux系统使用的是时间片算法,Windows则属于抢占式。这两种策略最终会带来不同的线程执行顺序;之前在推荐的一篇文章《》详细的介绍过,这里再简单的说一下;

时间片算法

在时间片算法中,所有的进程排成一个队列。操作系统按照他们的顺序,给每个进程分配一段时间,即该进程允许运行的时间。如果在时间片结束时进程还在运行,则CPU将被剥夺并分配给另一个进程。

如果进程在时间片结束前阻塞或结束,则CPU当即进行切换。调度程序所要做的就是维护一张就绪进程列表,当进程用完它的时间片后,它被移到队列的末尾。

抢占式操作系统

所谓抢占式操作系统,就是说如果一个进程得到了 CPU 时间,除非它自己放弃使用 CPU ,否则将完全霸占 CPU 。因此可以看出,在抢占式操作系统中,操作系统假设所有的进程都是“人品很好”的,会主动退出 CPU 。

在抢占式操作系统中,假设有若干进程,操作系统会根据他们的优先级、饥饿时间(已经多长时间没有使用过 CPU 了),给他们算出一个总的优先级来。操作系统就会把 CPU 交给总优先级最高的这个进程。

当进程执行完毕或者自己主动挂起后,操作系统就会重新计算一 次所有进程的总优先级,然后再挑一个优先级最高的把 CPU 控制权交给他。


相比下来,抢占式就有点像野蛮人;这也就解释了 Windows 为什么每次输出的结果都不太一样;由于测试用例执行的内容并不复杂,Linux 在最开始的时候,同一时间片下,就能执行完所有的指令,因此看起来就像线程是有序的在执行,当线程多导致同一时间片无法完成之后,同样也会出现错乱,但相比 Windows 更加有规律;

以下是 200 个线程的执行结果:

所以,不管是 Windows 还是 Linux,最终执行 run 方法的顺序跟线程调用start() 顺序并不是一致,不能通过 start() 方法来决定线程的执行顺序;跟 ReentrantLock 公平锁就更没有任何关系了;

Thread.run()

说了start() 方法,顺便说一下另外一个 run() 方法;

以下是 Thread.run() 方法的源码:

/**
 * If this thread was constructed using a separate
 * <code>Runnable</code> run object, then that
 * <code>Runnable</code> object's <code>run</code> method is called;
 * otherwise, this method does nothing and returns.
 * <p>
 * Subclasses of <code>Thread</code> should override this method.
 *
 * @see     #start()
 * @see     #stop()
 * @see     #Thread(ThreadGroup, Runnable, String)
 */
@Override
public void run() {
    if (target != null) {
        target.run();
    }
}

只有当线程创建时,传入了 Runnable对象,调用此方法才会有效,而且执行的过程并不是一个异步任务,不会交给“线程规划器”来管理,是由调用 run() 方法的主线程(main)在执行;通过 Thread.run() 执行的代码,是可以保证线程的执行顺序的,但是多线程的执行过程就由并行变成了串行化,也就是说,线程1执行完在执行线程2,线程2执行完才会轮到线程3 .... 一直到最后一个线程;这样就失去了多线程的意义了,因此, Thread.run() 在实际开发中要酌情使用。

线程的有序执行

上面的问题既然是线程执行顺序带来的,那如何让多线程有序的执行呢?

在多线程开发中,要求线程有序执行的场景不太多,如果同一个业务的多线程执行需要有序,那就间接性的将并行化变成了串行,似乎就没有使用多线程的必要了;但不排除不同功能,不同需求间,对执行顺序有要求,下面就列举几种让多线程有序执行的方法;

join()

join方法Thread类中的一个方法,作用是等待该线程执行直到终止。

public class ThreadTest {
    public static void main(String[] args) throws Exception{
        Thread t1 = new Thread(()-> System.out.println(Thread.currentThread().getName()+ "线程1"));
        Thread t2 = new Thread(()-> System.out.println(Thread.currentThread().getName()+ "线程2"));
        Thread t3 = new Thread(()-> System.out.println(Thread.currentThread().getName()+ "线程3"));
        t1.start();
        t1.join();
        t2.start();
        t2.join();
        t3.start();
    }
}

FutureTask

利用 get() 方法阻塞等待返回,只有执行完之后,才会继续往后执行;

public  static void futureTask() throws Exception{
    FutureTask<String> f1 = new FutureTask<>(() -> {
        System.out.println(Thread.currentThread().getName() + " FutureTask1");
        return "FutureTask1";
    });
    FutureTask<String> f2 = new FutureTask<>(() -> {
        System.out.println(Thread.currentThread().getName() + " FutureTask2");
        return "FutureTask2";
    });
    FutureTask<String> f3 = new FutureTask<>(() -> {
        System.out.println(Thread.currentThread().getName() + " FutureTask3");
        return "FutureTask3";
    });
    Thread t1 = new Thread(f1,"线程1");
    Thread t2 = new Thread(f2,"线程2");
    Thread t3 = new Thread(f3,"线程3");
    t1.start();
    System.out.println(f1.get());
    t2.start();
    System.out.println(f2.get());
    t3.start();
    System.out.println(f3.get());
}

单线程池

单线程池同一时间只允许有一个线程在执行,因此可以将要执行的线程按顺序提交给线程池,就能实现顺序执行的效果;

public static void singleThreadExecutor() {
    ExecutorService threadPool = Executors.newSingleThreadExecutor();
    Thread t1 = new Thread(() -> System.out.println(Thread.currentThread().getName() + " 线程1"));
    Thread t2 = new Thread(() -> System.out.println(Thread.currentThread().getName() + " 线程2"));
    Thread t3 = new Thread(() -> System.out.println(Thread.currentThread().getName() + " 线程3"));
    threadPool.execute(t1);
    threadPool.execute(t2);
    threadPool.execute(t3);
    threadPool.shutdown();
}

CountDownLatch

利用 CountDownLatch 的 await() 阻塞方法,后一个线程只有等前一个线程执行完并调用 countDown(),才能继续执行;

public static void countDownLatch() {
    CountDownLatch c1 = new CountDownLatch(1);
    CountDownLatch c2 = new CountDownLatch(1);
    Thread t1 = new Thread(() -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + " 线程1 等待1000ms");
        c1.countDown();
    });

    Thread t2 = new Thread(() -> {
        try {
            c1.await();
            System.out.println(Thread.currentThread().getName() + " 线程2");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        c2.countDown();
    });

    Thread t3 = new Thread(() -> {
        try {
            c2.await();
            System.out.println(Thread.currentThread().getName() + " 线程3");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
    t1.start();
    t2.start();
    t3.start();
}

好了,今天这个问题就说到这;希望本文能对你有所帮助...

码字不易,期待您的点赞、分享!