目录

1. 信号量

1.1 信号量和信号量操作的概念

1.2 信号量的基本使用接口

2. 基于环形队列的生产者消费者模型

2.1 环形队列再分析

2.2 代码分步实现

sem.hpp

ringQueue.hpp

testMain.cc

2.3 代码解析和再理解

3. 自旋锁和读写锁

3.1 自旋锁的概念和接口

3.2 读写锁的概念和接口

3.3 读写锁的原理和优先级

4. 笔试题

答案及解析

本篇完。

---

1. 信号量

之前在学习进程间通信的时候，简单地介绍过一下信号量，今天在这里进行详细的介绍。

    void push(const T& in) // 生产者
    {
        lockGuard lockgrard(&_mtx); // 自动调用构造函数
        //pthread_mutex_lock(&_mtx);

        // pthread_cond_wait: 只要是一个函数，就可能调用失败，可能存在 伪唤醒 的情况，所以用while
        while(isQueueFull()) //1. 先检测当前的临界资源是否能够满足访问条件
        {
            pthread_cond_wait(&_Full, &_mtx); // 满的时候就在_Full这个条件变量下等待
            // 此时思考：我们是在临界区中，我是持有锁的，如果我去等待了，锁该怎么办呢？
            // 所以pthread_cond_wait第二个参数是一个锁，当成功调用wait之后，传入的锁，会被自动释放
            // 当我被唤醒时，我从哪里醒来呢？->从哪里阻塞挂起，就从哪里唤醒, 被唤醒的时候，我们还是在临界区被唤醒的
            // 当我们被唤醒的时候，pthread_cond_wait，会自动帮助我们线程获取锁
        }

        _bq.push(in); // 2. 队列不为空或者被唤醒 -> 访问临界资源，100%确定，资源是就绪的

        pthread_cond_signal(&_Empty); // 唤醒
        // pthread_mutex_unlock(&_mtx); // 解锁
    } // 出了代码块自动调用析构函数

这是之前写的基于阻塞队列的生产者消费者模型中向阻塞队列中push任务的代码。

不足之处：

一个线程在向阻塞队列中push任务的时候，必须满足临界资源不满的条件，否则就会被放入到条件变量的等待队列中去。

但是临界资源是否为满是不能直接得到答案的，需要先申请锁，然后进入临界区访问临界资源去判断它是否为满。

在判断临界资源是否满足条件的过程中，必须先加锁，再检测，再操作，最后再解锁。

检测临界资源的本质也是在访问临界资源。

只要对临界资源整体加锁，就默认现场会对这个临界资源整体使用，但是实际情况可能存在：一份临界资源，划分为多个不同的区域，而且运行多个线程同时访问不同的区域。

• 在访问临界资源之前，无法得知临界资源的情况。
• 多个线程不能同时访问临界资源的不同区域。

---

1.1 信号量和信号量操作的概念

信号量：本质是一把计数器，用来衡量临界资源中资源数量多少。

申请信号量的本质：对临界资源中特定的小块资源的预定机制。

信号量也是一种互斥量，只要申请到信号量的线程，在未来一定能够拥有一份临界资源。

如上图所示，将一块临界资源划分为9个不同的区域，

现在想要让多个线程同时访问这9个不同的区域：

1. 创建一个信号量，它的值是9。
2. 每一个来访问临界资源的线程都先申请信号量，也就是将计数值减一。
3. 当计数值被减到0的时候，说明临界资源中的9个区域都有现场再访问，其他想要访问临界资源的现场只能阻塞等待。

申请到信号量的现场就可以进入临界区去访问临界资源，当访问完毕以后，再将信号量加一。

每个线程访问临界资源中的哪块区域由程序员决定，但是必须保证一个区域只能有一个现场在访问。

通过信号量的方式就解决了之前代码的不足：

• 线程不用访问临界资源就可以知道资源的使用情况。

信号量只要申请成功就一定有资源使用，只要申请失败就说明条件不满足，只能阻塞等待。

• 临界资源中的不同区域可以被多线程同时访问。

所有线程必须都能看到信号量才能申请，所以信号量是一个公共资源，公共资源就涉及到线程安全问题。

根据上面分析，信号量的基本操作就是对信号量进行加一和减一，所以这两个操作是原子的。

• P操作：就是信号量减减(sem–)，也就是在申请资源，而且该操作必须是原子的。
• V操作：就是信号量加加(sem++)，也就是在归还资源，同样也必须是原子的。

---

1.2 信号量的基本使用接口

#include <semaphore.h> // 信号量必须包含的头文件

sem_t sem; // 创建信号量

初始化信号量，man sem_init：

• sem：信号量指针
• shared：0表示线程间共享，非0表示进程间共享。我们一般情况下写0就行。
• value：信号量初始值，也就是计数器的值。
• 返回值：成功返回0，失败返回-1，并且设置errno。

---

信号量销毁，man sem_destroy：

• sem：信号量指针
• 返回值：成功返回0，失败返回-1，并且设置errno。

---

申请信号量（P操作 -> 计数器减减），man sem_wait：

• sem：信号量指针。
• 返回值：成功返回0，失败返回-1，并且设置errno。
• 功能：发布信号量，表示资源使用完毕，可以归还资源，将信号量值加1，也就是在归还信号量。

---

发布信号量（V操作 -> 计数器加加），man sem_post：

• sem：信号量指针。
• 返回值：成功返回0，失败返回-1，并且设置errno。
• 功能：发布信号量，表示资源使用完毕，可以归还资源，将信号量值加1，也就是在归还信号量。

---

这些接口和前面mutex的接口非常类似，因为他们都是POSIX标准的，所以使用起来没有难度。（以前简单讲的是SystemV标准的信号量）

---

2. 基于环形队列的生产者消费者模型

以前数据结构设计过环形队列（力扣622）：数据结构与算法⑨(第三章_下)队列的概念和实现(力扣：225+232+622)_GR_C的博客-CSDN博客

2.1 环形队列再分析

这里使用信号量来实现一个单生产单消费的环形队列模型。

• 环形队列采用数组来模拟，用取模运算来模拟环状特性。

环形结构起始状态和结束状态都是一样的，不好判断为空或者为满。

• 当环形队列为空时，头和尾都指向同一个位置。
• 当环形队列为满时，头和尾也都指向同一个位置。
• 可以通过加计数器或者标记位来判满或者空，也可以预留一个空的位置，作为满的状态。

但是我们现在有信号量这个计数器，就不需要用数据结构的方式来判空和判满了，能够很简单的进行多线程间的同步。

单生产者和单消费者一共两个线程在访问环形队列这个公共资源，生产者向环形队列中生产数据，消费者从环形队列中消费数据。

---

生产者和消费者什么情况下会访问同一个位置？ -> 环形队列为空和为满的时候

① 环形队列为空的时候，生产者和消费者会访问同一个位置。

当队列为空的时候，生产者访问队尾，向队列中生产数据，消费者访问对首消费数据，由于环形队列且为空，所以队首和队尾是同一个位置。

---

② 环形队列为满的时候，生产者和消费者会访问同一个位置。

当环形队列只有一个空位置的时候，生产者访问队尾生产数据，生产完毕后指向下一个位置，由于环形队列且为满，所以此时生产者又指向了队首，和消费者访问同一个位置。

其他任何时候，生产者和消费者访问的都是不同的区域。只要环形队列不满也不空，那么生产者和消费者之间都有数据，它们各自访问各自的区域。

---

为了完成环形队列的生产消费问题，必须要做的核心工作是什么？

① 消费者不能超过生产者。

消费者消费的是生产者生产的数据，生产者没有生产，消费者就无法消费。当消费者超过生产者后，消费者访问的区域并没有数据，所以没有任何意义。

消费者必须跟着生产者的后面，即使消费速度非常快（导致环形队列为空），此时消费者和生产者访问同一区域。

---

② 生产者不能把消费者套一圈以上

消费者消费的速度比较慢，环形队列满了以后，如果生产者继续生产，就会将消费者还没来得及消费的数据覆盖，消费者就无法消费到覆盖之前的数据了。

对于生产者而言，它在意的是环形队列中空闲的空间。

生产者只负责将数据生产到环形队列的空间中，当环形队列满了以后就不能生产了，所以它只关心环形队列中有多少空间可以用来生成数据。

对于消费者而言，它在意的是环形队列中 数据的个数。

消费者只负责从环形队列中消费数据，当环形队列为空时就停止消费，所以它只关心环形队列中有多是个数据可以用来消费。

• 空间资源定义一个信号量。用来统计空闲空间的个数。
• 数据资源定义一个信号量。用来统计数据个数。

所以生产者每次在访问临界资源之前，需要先申请空间资源的信号量，申请成功就可以进行生产，否则就阻塞等待。

消费者在访问临界资源之前，需要申请数据资源的信号量，申请成功就可以消费数据，否则就阻塞等待。

• 空间资源信号量的申请(P操作)由生产者进行，归还(V操作)由消费者进行，表示生产者可以生产数据。
• 数据资源信号量的申请(P操作)由消费者进行，归还(V操作)由生产者进行，表示消费者可以进行消费。

下面写写伪代码：

在信号量的初始化时，空间资源的信号量为环形队列的大小，因为没有生产任何数据。数据资源的信号量为0，因为没有任何数据可以消费。

通过信号量的方式同样维护了环形队列的核心操作，消费者消费速度快时，会将数据资源信号量全部申请完，但是此时生产者没有生产数据，也就没有归还数据资源的信号量，所以消费者会阻塞等待，不会超生产者。

生产者生产速度快时，会将空间资源信号量全部申请完，但是此时消费者没有消费数据，也就没有归还空间资源的信号量，所以生产者会阻塞等待，不会超过套消费者一个圈。

生产者伪代码：

productor_sem = 环形队列大小;

P(productor_sem);//申请空间资源信号量
//申请成功，继续向下运行。
//生气失败，阻塞在申请处。

.......//从事生产活动——把数据放入队列中

V(comsumer_sem);//归还数据资源信号量

消费者伪代码：

comsumer_sem = 0;

P(comsumer_sem);//申请数据资源信号量
//申请成功，继续向下运行。
//生气失败，阻塞在申请处。

.......//从事消费活动——从队列中消费数据

V(proudctor_sem);//归还空间资源信号量

在环形队列中，大部分情况下单生产和单消费是可以并发执行的，只有在满或者空的时候，才会有同步和互斥问题，同步和互斥是通过信号量来实现的。

在生产者和消费者并发访问环形队列时，访问的位置其实就是队列的下标，而且是两个下标。

当空或者满的时候，两个下标相同。

---

2.2 代码分步实现

先把代码架构敲出来：（和上一篇架构是一样的，只是“交易场所”从阻塞队列变成了环形队列）

Makefile：

ring_queue:testMain.cc
	g++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
	rm -f ring_queue

ringQueue.hpp：

#ifndef _Ring_QUEUE_HPP_
#define _Ring_QUEUE_HPP_

#include <iostream>
#include <vector>

const int g_default_num = 5; // 环形队列默认个数

template <class T>
class RingQueue
{
public:
    RingQueue(int default_num = g_default_num)
        : _ring_queue(default_num)
        , _num(default_num)
    {
    }
    ~RingQueue()
    {
    }
    void push(const T &in) // 生产者
    {
    }
    void pop(T *out) // 消费者
    {
    }
    void debug()
    {
        std::cerr << "size: " << _ring_queue.size() << " num: " << _num << std::endl;
    }

protected:
    std::vector<T> _ring_queue;
    int _num; // 环形队列的数据个数
};

testMain.cc：

#include "ringQueue.hpp"

void *consumer(void *args)
{
}

void *productor(void *args)
{
}

int main()
{
    RingQueue<int> *rq = new RingQueue<int>();
    rq->debug();

    // pthread_t c[3], p[2];
    // pthread_create(c, nullptr, consumer, (void *)rq);
    // pthread_create(c + 1, nullptr, consumer, (void *)rq);
    // pthread_create(c + 2, nullptr, consumer, (void *)rq);

    // pthread_create(p, nullptr, productor, (void *)rq);
    // pthread_create(p + 1, nullptr, productor, (void *)rq);

    // for (int i = 0; i < 3; i++)
    //     pthread_join(c[i], nullptr);
    // for (int i = 0; i < 2; i++)
    //     pthread_join(p[i], nullptr);

    return 0;
}

编译运行测试下：

符合预期。

---

下面实现一下push和pop，我们可以给push和pop都定义一个下标，定义成成员变量，这样想看看环形队列的结构还可以在debug打印出来。我们还要封装下信号量，下面放的就是完整代码了：

sem.hpp

#ifndef _SEM_HPP_
#define _SEM_HPP_

#include <iostream>
#include <semaphore.h>

class Sem
{
public:
    Sem(int value) // 传入的初始默认值
    {
        sem_init(&_sem, 0, value); // 0 -> 不需共享
    }
    void p() // P操作 -> 计数器减减 -> 申请信号量
    {
        sem_wait(&_sem);
    }
    void v() // V操作 -> 计数器加加 -> 发布信号量
    {
        sem_post(&_sem);
    }
    ~Sem() // 析构，直接销毁信号量
    {
        sem_destroy(&_sem); 
    }
protected:
    sem_t _sem; // 本质是计数器
};

#endif

ringQueue.hpp

#ifndef _Ring_QUEUE_HPP_
#define _Ring_QUEUE_HPP_

#include <iostream>
#include <vector>
#include <pthread.h>
#include "sem.hpp"

const int g_default_num = 5; // 环形队列默认个数

template <class T>
class RingQueue
{
public:
    RingQueue(int default_num = g_default_num)
        : _ring_queue(default_num),
          _num(default_num),
          c_step(0),
          p_step(0),
          _space_sem(default_num),
          _data_sem(0)
    {
        pthread_mutex_init(&clock, nullptr);
        pthread_mutex_init(&plock, nullptr);
    }
    ~RingQueue()
    {
        pthread_mutex_destroy(&clock);
        pthread_mutex_destroy(&plock);
    }
    void push(const T &in) // 生产者：关注空间资源
    {
        _space_sem.p(); // 先申请信号量 -> P操作（这样不用访问临界资源就分配好资源了）

        pthread_mutex_lock(&plock); 
        // 临界区：一定是竞争成功的生产者线程 -> 就一个
        _ring_queue[p_step++] = in;
        p_step %= _num; // p_step永远是可以存放的位置
        pthread_mutex_unlock(&plock);

        _data_sem.v();
    }
    void pop(T *out) // 消费者：关注数据资源 -> 对比生产者
    {
        _data_sem.p();

        pthread_mutex_lock(&clock);
        *out = _ring_queue[c_step++];
        c_step %= _num;
        pthread_mutex_unlock(&clock);

        _space_sem.v();
    }
    void debug()
    {
        std::cerr << "size: " << _ring_queue.size() << " num: " << _num << std::endl;
    }

protected:
    std::vector<T> _ring_queue;
    int _num; // 环形队列的数据个数
    int c_step; // 消费下标
    int p_step; // 生产下标
    Sem _space_sem;
    Sem _data_sem;
    pthread_mutex_t clock;
    pthread_mutex_t plock;
};

#endif

testMain.cc

#include "ringQueue.hpp"
#include <cstdlib>
#include <ctime>
#include <sys/types.h>
#include <unistd.h>

void *consumer(void *args)
{
    RingQueue<int> *rq = (RingQueue<int> *)args;
    while (true)
    {
        sleep(1);
        int x;
        rq->pop(&x); // 1. 从环形队列中获取任务或者数据
        // 2. 进行一定的处理 -- 不要忽略它的时间消耗问题
        std::cout << "消费: " << x << " [" << pthread_self() << "]" << std::endl;
    }
}

void *productor(void *args)
{
    RingQueue<int> *rq = (RingQueue<int> *)args;
    while (true)
    {
        // sleep(1);
        // 1. 构建数据或者任务对象 -- 一般是可以从外部来 -- 不要忽略它的时间消耗问题
        int x = rand() % 100 + 1;
        std::cout << "生产: " << x << " [" << pthread_self() << "]" << std::endl;

        // 2. 推送到环形队列中
        rq->push(x); // 完成生产的过程
    }
}

int main()
{
    srand((uint64_t)time(nullptr) ^ getpid());

    RingQueue<int> *rq = new RingQueue<int>();
    // rq->debug();

    pthread_t c[3], p[2];
    pthread_create(c, nullptr, consumer, (void *)rq);
    pthread_create(c + 1, nullptr, consumer, (void *)rq);
    pthread_create(c + 2, nullptr, consumer, (void *)rq);

    pthread_create(p, nullptr, productor, (void *)rq);
    pthread_create(p + 1, nullptr, productor, (void *)rq);

    for (int i = 0; i < 3; i++)
        pthread_join(c[i], nullptr);
    for (int i = 0; i < 2; i++)
        pthread_join(p[i], nullptr);

    return 0;
}

可以跟着注释看，编译运行：

---

2.3 代码解析和再理解

环形队列的生产者消费者模型同样遵循123原则：

• 1：一个交易场所，环形队列。
• 2：两种角色，生产者和消费者。
• 3：三种关系，生产者和生产者(互斥关系)，消费者和消费者(互斥关系)，生产者和消费者(在队列为空或者满时 -> 同步和互斥关系)。

上面的单生产单消费模型，维护的只是生产者和消费者之间的关系，要想实现多生产多消费，只需要将另外两种关系维护好即可。

•  在RingQueue中增加两把互斥锁，一把生产者使用，一把消费者使用。
•  在构造函数中将锁初始化，在析构函数中将锁摧毁。

push是向环形队列中生产任务，是生产者在调用，所以在生产之前需要加锁。pop是从环形队列中消费认为，是消费者在调用，所以在消费之前加锁。

互斥锁和申请信号量谁在前比较合适呢？

如果互斥锁在前，申请信号量在后：

• 所有生产者线程或者是消费者线程都需要先竞争锁，然后再去申请信号量，信号量申请成功才能进入临界区。
• 如果信号量申请失败就抱着锁阻塞，其他同类型线程就无法申请到锁。

这就好比去电影院买票，必须先排队进入放映厅才能买票。

如果申请信号量在前，互斥锁在后：

• 所有生产者线程或者消费者线程先申请信号量，再去申请锁，然后进入临界区。
• 如果信号量申请失败就不会再去申请锁。

同样是电影院，这就好比先买票，然后再排队进入放映厅，没买上票就没必要排队了。

对于线程来说，申请锁也是有代价的，将信号量申请放在前面可以减少申请锁的次数，所以申请信号量在互斥锁之前更合适。

创建多个生产者线程和多个消费者线程，去执行生产计算任务和消费计算任务。

• 生产任务的线程是不同的，可以根据tid值区别出来。
• 消费认为的现场也是不同的，同样可以根据tid值区别。

此时就实现了基于环形队列的多生产多消费模型。

多生产多消费的意义在哪里？

不要狭隘的认为，把任务或者数据放在交易场所，就是生产和消费了。

将数据或者任务生产前和拿到之后处理，才是最耗费时间的。

生产的本质：私有的任务-> 公共空间中

消费的本质：公共空间中的任务-> 私有的

信号量本质是一把计数器-> 计数器的意义是什么？

可以不用进入临界区，就可以得知资源情况，甚至可以减少临界区内部的判断

申请锁 -> 判断与访问 -> 释放锁 -> 本质是我们并不清楚临界资源的情况

信号量可以提前预设资源的情况，而且在pv变化过程中，我们可以在外部就能知晓临界资源的情况

---

3. 自旋锁和读写锁

之前使用的互斥锁就是挂起等待锁。多线程在竞争互斥锁时，申请到锁的线程进入临界区，而没有申请到锁的线程阻塞等待。

所谓的阻塞等待，其实是将该现场放入到操作系统维护的等待队列中，在合适的时候，操作系统再将其唤醒，放到运行队列中继续去申请锁。

其他常见的各种锁

• 悲观锁：在每次取数据时，总是担心数据会被其他线程修改，所以会在取数据前先加锁（读锁，写锁，行 锁等），当其他线程想要访问数据时，被阻塞挂起
• 乐观锁：每次取数据时候，总是乐观的认为数据不会被其他线程修改，因此不上锁。但是在更新数据前， 会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式：版本号机制和CAS操作。
• CAS操作：当需要更新数据时，判断当前内存值和之前取得的值是否相等。如果相等则用新值更新。若不等则失败，失败则重试，一般是一个自旋的过程，即不断重试。
• 自旋锁，公平锁，非公平锁？
• 读写锁。

穿越过来贴个了解知识：原子性原语double CAS (DCAS) 运行子两个随机排序内存单元上。若当前值与预期值一致，可改变这两个内存单元的值。

所谓原语的原子性操作是指一个操作中的所有动作，要么成功完成，要么全不做。也就是说，原语操作是一个不可分割的整体。为了保证原语操作的正确性，必须保证原语具有原子性。在单机环境下，操作的原子性一般是通过关闭中断来实现的。由于中断是计算机与外设通信的重要手段，关闭中断会对系统产生很大的影响，所以在实现时一定要避免原语操作花费时间过长，绝对不允许原语中出现死循环。  

这里简单的介绍下自旋锁和读写锁。

3.1 自旋锁的概念和接口

自旋锁也是互斥锁，它的作用也是保护共享资源的安全。多线程在竞争自旋锁时，申请到锁的线程进入临界区，而没有申请到锁的线程不会挂起等待。

没有申请到锁的线程会不停的继续去申请锁，直到申请锁成功进入临界资源，自旋和进程等待中的轮询非常的相似。

自旋锁和挂起等待锁的区别就在于：没有申请到锁时，自旋锁仍然继续申请，挂起等待锁则进入等待队列等待，在被唤醒后继续申请锁。

---

是什么决定着线程的等待方式呢？是需要等待的时长。

当访问临界资源的时间较短的时候，可以使用自旋锁，因为进入临界区的线程会非常快的出来，处于自旋状态的线程也可以很快进入临界区。

此时申请自旋锁的线程，免去了被挂起等待和唤醒的过程，一定程度上提高了效率。

当访问临界资源的时间较长的时候，就要使用挂起等待锁，因为进入临界区的现场不会很快出来。

此时将申请锁失败的线程挂起，就将CPU资源空闲了出来，如果不挂起而处于自旋状态，则CPU就一直被占用。

那么需要等待的时间长短是如何定义的呢？

像前面写的多线程抢票的代码，对于tickets的访问就可以使用自旋锁。

对于需要进行复杂运算，高IO，以及等待某些软件标志就位的情况就是用挂起等待所。

等待时间的长短并没有明确的定义，使用自旋锁还是挂起等待锁根据具体情况来觉得。最好的方式是分别测试两种锁，哪种效率高就用哪种。

---

看看自旋锁的基本使用接口：（和挂起等待锁的使用基本一样）

#include <pthread.h>//使用自旋锁要包含的头文件

pthread_spinlock_t lock;//创建自旋锁

man pthread_spin_init：

初始化自旋锁：

int pthread_spin_init(pthread_spinlock_t* lock, int shared);

• lock：自旋锁指针
• shared：0表示线程间共享，非0表示进程间共享，和信号量初始化中的shared一样。
• 返回值：成功返回0，失败返回-1。

销毁自旋锁：

int pthread_spin_destroy(pthread_spinlock_t* lock);

加锁和解锁：

• lock：都是自旋锁指针
• 返回值：都是成功返回0，失败返回-1。

这些接口和之前学习的挂起等待锁以及信号量的使用非常相似，只是换个函数名而已，因为它们遵循POSIX标准。

---

3.2 读写锁的概念和接口

读写锁主要使用在读者写者模型中，读者写者模型和生产者消费者模型很类似，也遵循123原则：

• 1：一个交易场所，任意类型的数据结构。
• 2：两种角色，读者和写者。
• 3：三种关系，写者和写者(互斥)，读者和写者(同步和互斥)，读者和读者(没有关系)。

读者线程和写者线程并发访问一块临界资源：

• 写者向临界资源中写数据。
• 读者从临界资源中读数据。

读者和写者之间是互斥关系
写者在写数据时，读者无法访问临界资源，因为如果在读取的时候，写者还没有写完，那么读者读到的数据就不全。

读者和写者之间也是同步关系
如果写者写好数据，读者不去都，那么写者写的数据就没有意义，所以写者写好数据后必须有读者来读。

反之，如果所有读者都已经读取过临界区的数据了，再读就是重复的旧数据，此时读取也没有意义，所以读者读完数据以后，写者必须来写入新的数据。

写者和写者直接是互斥关系
如果一个写者正在写数据，另一个写者也来写，假设他们写的是同一块公共资源，就有可能发生覆盖。

读者和读者之间没有关系
读者只从临界区中读取数据，并不拿走，所以读者之间并不会产生影响。

读者写者模型使用场景：一次发布，很长时间不做修改，大部分时间都是在被读取，比如这里写的博客。

---

读者写者模型和生产者消费者模型的本质区别是：消费者会拿走临界资源中的数据，而读者不会。

有些共享资源的数据修改的机会比较少，相比较改写，它们读的机会反而高的多。

在读的过程中，往往伴随着查找的操作，中间耗时很长。给这种代码段加锁，会极大地降低我们程序的效率。

读写锁就是专门用于读者写者模型中的一种锁，可以给读者加锁，也可以给写者加锁，可以维护读者写者的321原则。

临界区的状态	读者请求	写者请求
无锁	可以	可以
读锁	可以	阻塞
写锁	阻塞	阻塞

持有写锁的线程独占临界资源，持有读锁的线程，读者之间共享临界资源。

---

读写锁基本接口：（还是和以前用的差不多）

#include <pthread.h>//读写锁必须包含的头文件

pthread_rwlock_t rwlock;//创建读写锁

初始化读写锁：man pthread_rwlock_init

int pthread_rwlock_init(pthread_rwlock_t* rwlock, const pthread_rwlockattrt_t* attr);

• rwlock：读写锁指针
• attr：读写锁属性结构体指针，一般设置成nullptr即可。
• 返回值：成功返回0，失败返回-1

销毁读写锁：

int pthread_rwlock_destroy(pthread_rwlock_t* rwlock);

加读锁：

int pthread_rwlock_rdlock(pthread_rwlock_t* rwlock);

加写锁：

int pthread_rwlock_wrlock(pthread_rwlock_t* rwlock);

解锁：

int pthread_rwlock_unlock(pthread_rwlock_t* rwlock);

• 读锁和写锁都通过这个接口去解锁。

同样是POSIX标准，所以返回值，参数等风格和前面的挂起等待锁，信号量，以及自旋锁一样，这里就不详细解释了。

---

3.3 读写锁的原理和优先级

读写锁：在任何时刻，只允许一个写者写入，但是允许多个读者并发读取(写者阻塞)。

是不是感觉非常奇怪？上面的接口中明明只有一把锁，但是可以给读者和写者分别加锁，而且对于读者和写者的效果还不同？

下面写一段伪代码来解释一下：

读写锁的类型是一个结构体，它里面封装的也是互斥锁，而且针对读者有一把，针对写者有一把，只是机制不一样而已

读加锁伪代码： pthread_rwlock_rdlock(pthread_rwlock_t* rwlock)

pthread_mutex_t rdlock;//创建读锁
int reader_count = 0;//读者计数
------------------------------------------------------------
lock(&rdlock);//读加锁
reader_count++;//读者数量加一
if(reader_count == 1)
{
	//只要有读者在访问临界资源，就将写锁也申请走
	lock(&rwlock);//写加锁
}
unlock(&rdlock);//解读锁
------------------------------------------------------------
//读取数据....
------------------------------------------------------------
lock(&rdlock);//再次读加锁
read_count--;//读者数量减一
if(reader_count == 0)
{
	//读者全部读完以后，释放写锁
	unlock(&rwlock);//写解锁
}
unlock(&rdlock);//读解锁

加读锁时，有一个计数器，该计数器所有读者线程共享，是一份共享资源，用来统计访问公共资源的读者数量。

伪代码解释：

每个读者访问公共资源的时候，都需要将计数值加1，考虑到线程安全，所以计数值要加锁。
当第一个读者到来后，它先申请了读锁，然后又申请了写锁，此时写者线程就无法访问临界资源了，因为写锁在读者手里。之后的读者线程仅将计数值加一即可。
当读者线程访问完计数值以后就将读锁解锁，然后去公共资源中读数据(仅读取，不拿走)。
读者读完数据以后，继续线程安全的访问计数值，将值减一，当值被减到0时，说明没有读者再来读数据了，此时将申请的写锁解锁，好方便写者访问公共资源。

通过这样的方式就实现了读者和写者之前的互斥，读者和读者之间没有关系。

互斥访问读者计数值非常的快，读者真正访问公共资源的时候是没有任何关系的（不存在加锁）。

写加锁伪代码： pthread_rwlock_wrlock(pthread_rwlock_t* lock)

pthread_mutex_t wrlock;//创建写锁
------------------------------------------------------------
lock(&wrlock);//写加锁
//向临界资源中写入数据
unlock(&wrlock);//写解锁

写者的加锁解锁，实现了写者之间的互斥关系。

伪代码解释：

写者线程在访问临界资源的时候会先申请锁，申请成功的进入临界区，失败的阻塞等待。
如果写者申请写锁成功，那么第一个读者在申请写锁的时候同样会阻塞，直到写者释放锁。
如果第一个读者申请写锁成功，那么写者在申请写锁的时候也会阻塞，直到读者释放锁。

写锁的原理非常简单，正是由于读者会申请写锁，写者也会申请写锁，所以才能实现写者和读者的互斥。

---

上面讲解的读写锁是读者优先的，前提是有读者已经在访问公共资源。

已经有读者在访问公共资源的时候，写锁已经被读者申请走了。

当后面写者和读者同时到来的时候，写者会因为无法申请锁而阻塞，而读者可以访问公共资源。

如果没有读者在访问公共资源，第一个读者和写者同时到来时，它两就不存在优先关系，谁的竞争能力强谁申请到写锁，进入临界资源。

试想，读者非常多，那么写者就始终无法进入临界区访问临界资源，所以就会导致写者饥饿问题，但是读写锁就是应用在这种场景下，写者数量少执行少，读者数量多执行多。

读写锁是可以设置成写者优先的。

即使已经有读者在访问公共资源，并且写锁已经被申请走了。

当后面的写者和读者同时到来的时候，将写者后面的所有读者阻塞，不让它们访问公共资源。

当进入临界区的读者出来以后，并且归还了写锁，此时写者直接申请写锁并进入临界区访问临界资源。

大概的道理是这样，具体如何阻塞写者之后的读者策略可以在代码层面进行设计。

pthread库其实是提供了设置读写优先级的接口的：

int pthread_rwlockattr_setkind_np(pthread_rwlockattr_t* attr, int pref);

• attr：属性设置
• pref：有三种选择
• PTHREAD_RWLOCK_PREFER_READER_NP：(默认设置)读者优先，可能会导致写者饥饿情况。
• PTHREAD_RWLOCK_PREFER_WRITER_NP：写者优先
• PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP：写者优先，但写者不能递归加锁。

---

4. 笔试题

1. 下面有关进程间同步的几种方式的区别，描述正确的是？ [多选] 

A.因为使用互斥不仅仅能够在同一应用程序不同线程中实现资源的安全共享，而且可以在不同应用程序的线程之间实现对资源的安全共享。

B.任意时刻只允许一个线程对共享资源进行访问。如果有多个线程试图同时访问临界区，那么 在有一个线程进入后其他所有试图访问此临界区的线程将被挂起，并一直持续到进入临界区的线程离开

C.信号量允许多个线程同时使用共享资源

D.如果只为了在进程内部使用的话使用临界区会带来速度上的优势并能够减少资源占用量

2. 以下哪几种方式可用来实现线程间通知和唤醒：( ) [多选] 

A.互斥锁

B.条件变量

C.信号量

D.读写锁

3. 信号量实现与条件变量有什么区别 [多选]

A.信号量既可以实现同步还可以实现互斥

B.条件变量既可以实现同步还可以实现互斥

C.条件变量需要搭配互斥锁使用，信号量不需要

D.信号量需要搭配互斥锁使用，条件变量不需要

4. 关于条件变量以下描述正确的有 [多选]

A.条件变量可以单独使用

B.条件变量需要搭配互斥锁使用

C.条件变量在等待被唤醒时需要重新对条件进行判断，是否条件满足

D.在生产者与消费者模型中只需要一个条件变量就可以

5. 已知如下代码，并在两个线程中同时执行f1和f2，待两个函数都返回后，a的所有可能值是哪些?[多选] 

    int a = 2, b = 0, c = 0
    void f1()             
    {                 
      b = a * 2;            
      a = b;               
    }
    void f2()
    {
      c = a + 11;
      a = c;
    }

A.4

B.13

C.15

D.26

6.  关于函数的重入于不可重入描述正确的是：

A.线程是安全的则线程中调用的函数一定是可重入函数

B.常见可重入的情况包括：不使用全局变量或静态变量，但是可以调用malloc/free等函数

C.函数可重入只是线程安全的一个要素

D.函数线程安全只是函数可重入的一个要素

7. 简述什么是线程同步，为什么需要同步

8. 简述线程安全概念与实现

---

答案及解析

1. ABCD

• A正确，与谁能互斥主要取决于互斥锁共享于哪些执行流之间，如果互斥锁为进程内的资源，则可以实现同一程序内的不同线程间互斥，而如果将共享内存作为互斥锁进行操作则可以实现不同进程之间的互斥
• B正确，临界资源的访问操作需要加锁保护，不能同时访问，否则有可能会造成数据二义，因此如果有多个线程试图同时访问，则必然需要通过互斥的方式保证同一时间只有一个执行流能访问，其他线程阻塞
• C正确，信号量主要用于实现同步操作，只要资源数大于0就表示可获取，可访问，因此上课时讲到若要使用信号量模拟实现互斥，则需要初始化资源计数为1，表示资源只有一个，则只有一个执行流能访问
• D正确，之前博客的临界区指的是访问临界资源的代码片段，而这里说的临界区指的是其他平台或语言中的一种锁技术，实现串行化来访问共享资源的代码片段。速度比较快但是只能用于同一进程的线程间

2. BC

线程间的通知和唤醒以及线程的等待这是线程间同步实现的基础，而信号量和条件变量通过提供的使线程等待和唤醒功能被用于实现线程间的同步，因此选择B和C

而A中的互斥锁，和D中的读写锁都是为了实现对共享资源安全访问操作的锁技术，并不包含有通知和唤醒线程的功能。

3. AC

• 条件变量提供了一个pcb阻塞队列以及阻塞和唤醒线程的接口用于实现同步，但是什么时候该唤醒以及什么时候该阻塞线程由程序员进行控制，而这个控制通常需要一个共享资源的条件判断完成，因此条件变量还需要搭配互斥锁使用，来保护这个共享资源的条件判断及操作。
• 信号量提供一个pcb等待队列，以及一个实现了原子操作的对资源进行计数的计数器，通过自身计数器实现同步的条件判断，因此不需要搭配互斥锁使用，而且信号量在初始化计数为1的情况下也可以模拟实现互斥操作。

4. BC

A错误 条件变量进行同步的条件判断由外部的共享资源条件判断实现，因此需要搭配互斥锁使用

B正确

C正确 条件变量的控制判断需要使用循环进行，避免在多个线程同时被唤醒的情况下，A线程加锁成功访问资源，其他线程卡在锁处，而A线程一旦解锁，其他线程抢到锁在资源访问条件不满足的情况下访问资源，因此被唤醒后加锁成功则需要重新进行判断，条件满足则访问，不满足则需要重新陷入休眠。

D错误 条件变量的使用中不同的角色需要等待在不同的条件变量等待队列中，防止角色误唤醒，比如生产者唤醒生产者的情况，因此需要分开等待，分开唤醒

5. ABCD

因为a，b，c变量是全局变量，因此在不同的线程中调用f1和f2函数有可能会出现竞态执行同时对三个变量进行操作的情况

A正确 存在f1函数被执行 b=a*2后b等于4， 然后时间片轮转到f2函数被执行完毕，时间片轮转到 f1中的a=b，则这时候a的值就可能是最后的赋值为4

B正确 跟上边类似，有可能f2韩式中的 c=a+11，也就是c=13执行完后， 时间片轮转到f1函数的执行，等到执行完毕后，轮转回来执行 a=c则最终赋值为13

C正确 两个函数串行执行，执行完f1之后a是4， 然后执行f2函数，c=a+11=15, 然后a=c则赋值为15

D正确 先执行了f2函数，则a=13， 然后时间片轮转到 f1函数的执行， b=a*2, a=b则最后赋值为26

6. C

A错误 线程安全指的是当前线程中对各项操作时安全的，但不表示内部调用的函数是安全的，两个之间并没有必然关系

B错误 调用了malloc/free函数，因为malloc函数是用全局链表来管理堆的，所以是不可重入的。

C正确 线程中不仅仅会调用函数，有可能本身内部就进行了临界资源的操作，所以线程内调用的函数可重入只是线程安全的一个要素

D错误 一个函数一旦是线程安全的，则表示在多个线程内重入不会引发意外问题，因此也是可重入的

7. 简述什么是线程同步，为什么需要同步

线程同步指的是线程间对数据资源进行获取，有可能在不满足访问资源条件的情况下访问资源而造成程序逻辑混乱，因此通过进行条件判断来决定线程在不能访问资源时休眠等待或满足资源后唤醒等待的线程的方式实现对资源访问的合理性

8. 简述线程安全概念与实现

线程安全指的是在多线程编程中，多个线程对临界资源进行争抢访问而不会造成数据二义或程序逻辑混乱的情况。

线程安全的实现，通过同步与互斥实现

具体同步的实现可以通过互斥锁和信号量实现、而同步可以通过条件变量与信号量实现。

---

本篇完。

至此已经把常用的锁介绍完了，锁的种类非常的多，但是常用的就这几种，尤其是是mutex(挂起等待锁)，cond(条件变量)，sem(信号量)，以及基于阻塞队列和环形队列的生产者消费者模型尤为重要。

至于自旋锁和读写锁，使用的频率没有那么高，所以这里也没有去演示具体的代码，有兴趣的可以去自行尝试。

下一篇：零基础Linux_26（多线程）线程池代码+单例模式+线程安全。