操作系统（六）—— 线程安全

概述

　　线程安全问题，是一个非常重要且复杂的问题。在多线程修改共享变量的时候，如果不做特殊的处理，变量的最终的结果往往不会符合我们的预期，原因很简单，因为现代的计算机都做了多级的缓存，而且使用了多核，那每个线程修改某个变量是在CPU的高速缓存中修改，而且修改完之后并不是立即就同步到主存中，这就导致某一个线程已经修改了数据，但是主存并没有修改，别的线程要修改这个数据的时候，从主存再读取这个数据就是一个错误的结果。那本篇文章就介绍一下如何解决这个问题，在《现代操作系统》那本书中介绍了两种方法，一种是软件的方法，一种是硬件的方法，由于软件的方法实现起来很复杂，而且使用场景有限，本文就不做介绍了，只介绍硬件的实现方式。

竟争条件

两个或多个进程或线程读写某些共享数据，而最后的结果取决于进程运行的精确时序，就叫做竞争条件，举一个简单的例子，使用的是java代码。

public class Test {
    static int i = 0;

    private static final CountDownLatch latch=new CountDownLatch(20);

    public static void main(String[] args) {
        multiThread();
    }

    public static void multiThread(){

        for (int j = 0; j < 20; j++) {
            new Thread(new RunableImplements()).start();
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(i);


    }

    public static class RunableImplements implements Runnable{

        @Override
        public void run() {
            for (int j = 0; j < 5000; j++) {
                i++;
            }
            latch.countDown();
        }
    }
}

大家可以运行一下这个程序，最后的结果很大可能性不是100000。

做过java开发的朋友一眼应该就可以看明白，没有写过java的朋友，对CountDownLatch 可能不太熟悉，解释一下这个，CountDownLatch的作用就是一个计数器，初始值大小为线程大小，也就是20，每个线程执行完之后这个值就会减1，直到所有的线程执行完成，CountDownLatch的值也就变成了0，代码中的latch.await()方法就是等待所有的线程执行完。

上面的代码很好的体现了竞争条件。

原子操作 

一次不存在任何中断或者失败的执行，这是清华大学公开课给的定义，但是有点不太准确，准确的说，一个由多步操作组成的一个操作，原子操作是指要么执行完所有步骤，要么一步也不执行，不可能只执行所有步骤的一个子集。

临界区

对共享内存进行访问的代码片段叫做临界区。如果可以防止两个进程或者线程同时进入临界区，就可以解决竞争条件的问题。单单解决了竞争条件的问题，并不能保证使用共享数据并发的进程或线程能够正确且高效的协作，对于一个好的解决方案，要满足下面四个条件：

　　　　来源：现代操作系统

解决方案一：屏蔽中断

对于单核CPU来说，最简单的解决办法就是屏蔽中断，因为单核CPU并不存在真正的并发，多线程或者进程实际上还是交替执行的，当屏蔽中断时时钟中断也会屏蔽，CPU只有在发生时钟中断或者其他中断时才会出现对进程进行上下文切换，所以一旦屏蔽中断之后就只有自己这个进程进来，而不必担心别的进程会修改共享数据。

时钟中断解释：

linux是分时操作系统，就是CPU时间会分为多个时间片，比如10毫秒一个时间片，程序执行一个时间片之后，操作系统会重新选择一个任务来执行。问题是CPU是怎么知道时间片到了呢？又是如何触发任务选择的呢？

关键原理就是CPU有个外部时钟，这是一个倒数计时器，初始时会设置一个数字，比如1000，然后每个时钟脉冲数字减一，减到0的时候，就给CPU发一个信号，CPU会中断当前程序，来处理这个信号，这个信号的处理程序会重置计时器，并执行信号处理函数，如此反复，起到了时间分片的效果。（来源：知乎）

缺点：

1. 只能解决单个CPU的计算机，如果是多核就不管用了，因为屏蔽中断是屏蔽一个核的。其他的CPU依然可以访问共享数据。
2. 把中断的权限交给一个用户进程是很不明智的，假如这个用户进程开始执行，但是一直都不结束，这时候由于CPU无法做上下文切换，别的进程都无法执行。

解决方案二：基于硬件的原子指令

 硬件提供一些原语，比如中断禁用，原子指令等。常见的如下：

• TSL(test and set lock),测试并加锁
• XCHG,原子性的交换两个位置的值
• 信号量（semaphore）

这三种是硬件实现的原子操作，在这些硬件提供的原子操作的基础上，又提出了对这些原子操作的抽象，比如互斥量，管程。下面就先逐个介绍这些硬件原子操作的原理，介绍完之后，再介绍对他们的抽象，最后举一个例子来说明这些抽象的应用。

TSL

大家用自己的打字软件，输入tsl，是什么呢？O(∩_∩)O~，我的第一个结果竟然是特斯拉，一个天才发明家，一个天生就接近开悟的人，慧根很深。。。哦，扯远了，哈哈哈。

工作流程

将一个内存字lock读到寄存器RX中，之后在内存地址上存一个非0值，上面两步的读和写是一个原子操作，即在执行上面的过程中不允许别的进程对该内存字进程操作。具体的指令如下

　　　　　　　　　　　　　　图片来源：现代操作系统

 上面的过程中，第一步是把内存中的lock复制到寄存器中，然后将内存中的lock设置为1，之后判断寄存器中的值是否为0，如果为0，就加锁成功，当访问临界区结束，直接调用leave_region，将内存中lock设置为0。如果是1，说明已经有别的进程加锁成功了，回到初始状态，再次重试，可以看出TSL是忙等待的方式实现的，而且可以适用多个进程同时竞争锁。

实现原理

实现原理和上面介绍的屏蔽中断不同，TSL的原理是执行TSL的CPU通过锁住内存总线，在TSL操作结束之前不允许别的CPU访问该内存指令实现的。

XCHG

这个和上面介绍的TSL一样，通过原子的交换两个位置的值，比如寄存器和主存。下面看一下工作流程。

工作流程

 　　　　　　来源：现代操作系统

第一步先在寄存器存入1，之后和主存lock交换，之后判断寄存器中的值是不是0，如果是0，说明主存中没有别的进程加锁，加锁成功。否则返回到初始状态，继续重试，也是采用忙等待的方式。可以看出上面的过程基本上TSL一样，只是TSL没有交换，而是直接将内存中的lock复制到寄存器中，之后再把主存的值修改为1.

在itel x86 CPU中都是采用这种方式进行底层同步。

信号量

信号量是E.W.Dijkstra在1965年提出的，Dijkstra是计算机学界的一个巨佬，提出过非常多牛逼的设计，而且是图灵奖获得者,这些都不是重点，重点是这么牛逼的大佬，竟然没有秃。

                                         

　　　　　　　　　　　　　    E.W.Dijkstra（1930.5.11~2002.8.6）　

 工作流程

信号量用一个整形数字记录唤醒次数，取值范围为大于等于0的数字。信号量可以做两种操作，up操作和down操作，up操作是对信号量的值加1，down操作是对信号量的值减1，如果检查发现信号量的值为0，就不能减1了，该进程会进入睡眠状态，其中检测信号量的值，修改，睡眠等一系列的操作是一个原子操作。当没有执行down操作的进程时（比如都睡眠了），执行up操作的进程可以唤醒一个睡眠的进程来执行down操作，没有进程会因为up操作而进入睡眠。

实现原理

信号量可以正确的工作依赖两点。

第一、在单个CPU上，在进行信号量的值的检查，修改，睡眠进程等操作时会屏蔽中断，因为修改信号量等操作是一个很简单的操作，只有几条指令，可以很快执行完，使用屏蔽中断并不会影响到别的进程执行。这样就可以保证在单个CPU上信号量的操作是一个原子操作。

第二、在多个CPU上，信号量需要借助TSL或者XCHG等锁住内存总线，保证一次只有一个CPU可以对信号量操作，由于对于信号量的操作很快，TSL和XCHG即便是忙等待，也不会等待太久，而过多的消耗CPU性能。

二元信号量

信号量的值只有两个0或者1，这样就可以实现互斥。比如设置一个信号量的初值为1，每个想要进入临界区的进程需要执行一个down操作，如果信号量值是1，那可以down成功，否则就要睡眠，当某个进程访问临界区结束，执行一个up操作，这个时候唤醒别的等待的进程。

互斥量

这个其实和上面讲的二元信号量类似，也是只有两个值，比如0表示解锁状态，1表示加锁状态，来实现互斥，底层原理也是使用TSL或者XCHG，就不多介绍了。

管程

管程是一种更高级的抽象，也是为了完成互斥和根据特定条件阻塞进程。上面既然已经有了互斥量和信号量，为什么还要搞一个管程呢？因为信号量编程非常不方便，而且很容易出错，所以就又发明了一个抽象程度更高的东东。

管程解决的最主要的问题就是分离互斥和根据特定条件阻塞。

管程构成

一、互斥锁：任意时刻只允许一个进程或者线程进入临界区

二、一个或多个条件变量：条件变量包含两个操作等待 / 唤醒操作

小结

上面把各种实现同步互斥的工具都介绍了一遍，这里想单独提一下管程，做过java开发的胖友应该都使用过reentrantlock + condition组合，这个其实就是管程，其中reentrantlock实现互斥，condition实现条件状态。ok，上面介绍了那么多概念，看的估计很晕，下面就举一个例子，分别使用信号量和管程解决，应该就可以很好的理解了。

问题：生产者消费者问题

一个固定大小的缓存区，一个生产者向里面写数据，一个消费者消费数据（也可以是n个生产者，m个消费者，但是那样问题就相对复杂），当缓冲区满了之后，生产者阻塞，当缓冲区没有数据了，消费者阻塞。

使用信号量解决

使用三个信号量，第一个称为full,用来记录充满的缓冲槽数目。第二个称为empty,记录空的缓冲槽的数目。第三个称为mutex，是一个二元信号量，用来做互斥使用的，保证两个线程不能同时访问缓冲槽。代码如下

　　　　

 代码很简单，下面就分析一下生产者的代码。

1. 首先使用死循环，不停的写入
2. 当要写入的时候，先将空的槽减1，如果没有空槽位，阻塞
3. down(&mutex)，加锁
4. insert_item,将数据插入缓冲区
5. up(&mutem)，解锁
6. up(&full)，插入成功，充满的槽位加1

大家可以思考一个问题，第二步和第三步可不可以交换一下顺序？很多胖友可能是下面的回复

                           

 但是答案是不行，考虑下面一种情况，如果现在缓冲区已经满了，那empty = 0,这个时候正确的做法是生产者阻塞，让消费者去消费。但是如果把那两步交换，会出现一种情况就是，生产者执行down(&mutex)占用着锁，但是down(&empty)却阻塞在那儿了，当消费者进来获取锁时，获取不到锁，导致消费者也阻塞，这就是所谓的死锁状态，所以在上面介绍管程的时候说信号量对于编程来说并不友好，稍不留神，就会出现死锁。

使用管程解决

该代码是使用java实现的，使用的是java中的synchronized关键字，使用这个关键字实现互斥，然后使用wait()和notify()表示条件状态。具体代码如下

        

 别看这个代码写的很长，其实很简单，一个生产者的类，一个消费者的类，一个管程类，其中生产者和消费者类采用异步执行。生产者进来之后，要判断当前缓冲区是不是满了，缓冲区大小为N = 100，如果满了，就调用go_to_sleep()，执行wait()进行等待。如果没有满就正常插入。由于our_monitor是一个静态内部类，insert方法和remove方法都使用synchronized修饰，所以这两个方法不能同时执行（如果对这个有疑问的胖友可以执行一下我的下面这段代码，我也对这个有疑问，所以写了一个测试代码）。

public class Test01 {

    static CountDownLatch latch = new CountDownLatch(20);

    public static void main(String[] args) {
        Test02 test02 = new Test02();
        for (int i = 0; i < 10; i++) {
            new Thread(()-> test02.test1()).start();
            new Thread(()->test02.test2()).start();
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
    static class Test02{
        public synchronized void test1(){
            try{
                System.out.println("test1 come in");
                Thread.sleep(2000);
                System.out.println("test1 out");
                latch.countDown();
            }catch (Exception e){

            }

        }
        public synchronized void test2(){
            try {
                System.out.println("test2 come in");
                Thread.sleep(2000);
                System.out.println("test2 out");
                latch.countDown();
            }catch (Exception e){
            }
            
        }
    }
}

总结

本文主要介绍了多进程或者多线程并发的时候如何解决线程安全的问题，介绍了使用硬件实现的方案，并且介绍了比硬件实现更高级的抽象，管程和互斥量，最后介绍了一个经典的问题，生产者消费者问题，然后使用信号量和管程分别实现了一下。文章中直接截取《现代操作系统》书中的代码，有很详细的注释，相信大家应该可以很容易明白。

下一篇讲解进程间通信，我都想不到会有这么多东西，如同傻狗一般惊了。。。

                                            

参考：

《现代操作系统》

《程序员的自我修养》