进程同步经典示例 多线程上篇（五）

程序员潇然

同步回顾

进程同步控制有多种方式：算法、硬件、信号量、管程 这些方式可以认为就是同步的工具（方法、函数） 比如信号量机制中的wait(S) 和 signal(S) ，就相当于是两个方法调用。 调用wait(S)就会申请这个资源，否则就会等待（进入等待队列）；调用signal(S)就会释放资源（或一并唤醒等待队列中的某个）； 在梳理同步问题的解决思路时，只需要合理安排方法调用即可，底层的实现细节不需要关注。 接下来以这种套路，看一下借助与不同的同步方式“算法、硬件、信号量、管程”这一“API”，如何解决经典的进程同步问题

生产者消费者

生产者-消费者(producer-consumer)问题是一个著名的进程同步问题。它描述的是： 有一群生产者进程在生产产品，并将这些产品提供给消费者进程去消费。 为使生产者进程与消费者进程能并发执行，在两者之间设置了一个具有 n 个缓冲区的缓冲池，生产者进程将它所生产的产品放入一个缓冲区中；消费者进程可从一个缓冲区中取走产品去消费。 尽管所有的生产者进程和消费者进程都是以异步方式运行的，但它们之间必须保持同步 也就是即不允许消费者进程到一个空缓冲区去取产品，也不允许生产者进程向一个已装满产品且尚未被取走的缓冲区中投放产品。

记录型信号量

对于缓冲池本身，可以借助一个互斥信号量mutex实现各个进程对缓冲池的互斥使用； 生产者关注于缓冲池空位子的个数，消费者关注的是缓冲池中被放置好产品的满的个数 所以，我们总共设置三个信号量semaphore：

mutex值为1，用于进程间互斥访问缓冲池 full表示缓冲区这一排坑中被放置产品的个数，初始时为0 empty表示缓冲区中空位子的个数，初始时为n

对于缓冲池以一个数组的形式进行描述：buffer[n] 另外还需要定义两个用于对数组进行访问的下标 in 和 out ，初始时都是0，也就是生产者会往0号位置放置元素，消费者会从0号开始取 每次的操作之后，下标后移，in和out采用自增的方式，所以应该是循环设置，比如in为10时，应该从头再来，所以求余（简言之in out序号一直自增，通过求余循环）

//变量定义
int in=0, out=0;
item buffer[n];
semaphore mutex=l，empty=n, full=0;

//生产者
void proceducer(){
do{
  producer an item nextp;
  ......
  wait(empty);//等待空位子
  wait(mutex);//等待缓冲池可用
  buffer[in] =nextp;//设置元素
  in =(in+1)%n;//下标后移
  signal(mutex);//释放缓冲池
  signal(full);//“满”也就是已生产产品个数释放1个(+1)
}while(TRUE);


//消费者
void consumer() {
  do{
    wait(full);//等待已生产资源个数
    wait(mutex);//等待缓冲池可用
    nextc= buffer[out];//获得一个元素
    out =(out+1) % n;//下标后移
    signal(mutex);//释放缓冲池
    signal(empty);//空位子多出来一个
    consumer the item in nextc;//消费掉获得的产品
    } while(TRUE);
}

//主程序
void main() {
  proceducer();
  consumer();
}

以上就是一个记录型信号量解决生产者消费者的问题的思路 对于信号量中用于实现互斥的wait和signal必须是成对出现的，尽管他们可能位于不同的程序中，这都无所谓，他们使用信号量作为纽带进行联系

AND型信号量

对于生产者和消费者，都涉及两种资源，一个是缓冲池，一个是缓冲池空或满 所以可以将上面两种资源申请的步骤转换为AND型，比如 wait(empty);//等待空位子 wait(mutex);//等待缓冲池可用 转换为AND的形式的Swait(empty，mutex)

int in=0, out=0;
item buffer[n];
semaphore mutex=l, empty=n, full=O;
    void proceducer() {
    do{
    producer an item nextp;
    ......
    Swait(empty, mutex);
    buffer[in] = nextp;
    in =(in+1) % n;
    Ssignal(mutex, full)
    } while(TRUE);
}
void consumer() {
do{
    Swait(full, mutex);
    nextc= buffer[out];
    out =(out+1) % n;
    Ssignal(mutex, empty);
    consumer the item in nextc;
    ......
  } while(TRUE);
}

这个示例中，AND型信号量方案只是记录型信号量机制的一个简单升级

管程方案

管程由一组共享数据结构以及过程，还有条件变量组成。 共享的数据结构就是缓冲池，大小为n 生产者向缓冲池中放入产品，定义过程put(item) 消费者从缓冲池中取出产品，定义过程get(item) 对于生产者，非满 not full 就可以继续生产数据； 对于消费者，非空 not empty 就可以继续消费数据； 所以设置两个条件：notfull，notempty 如果数据个数 count>=N，那么 notfull 非满条件不成立 如果数据个数 count<=0，那么notempty 非空条件不成立 也就是说： count>=N，notfull 不满足，生产者就会在 notfull 条件上等待 count<=0N，notempty 不满足，消费者就会在 notempty 条件上等待

//定义一个管程
Monitor procducerconsumer {
item buffer[N];//缓冲区大小
int in, out;//访问下标
condition notfull, notempty;//条件变量
int count;//已生产产品的个数
//生产方法
void put(item x) {
  if(count>=N){
    notfull.wait; //如果生产个数已经大于缓冲区大小，将生产进程添加到notfull条件的等待队列中
  }
  buffer[in] = x; //设置元素
  in = (in+1) % N; //下标移动
  count++;//已生产产品个数+1
  notempty.signal //释放等待notempty条件的进程
}

//获取方法
void get(item x) {
  if(count<=0){
    notempty.wait; // 如果已生产产品数量为0（以下），消费者进程添加到notempty的等待队列中
  }
  x = buffer[out];// 读取元素
  out = (out+1) % N; // 下标移动
  count--; //已生产产品个数-1
  notfull.signal; // 释放等待notfull条件的进程
}
//初始化数据方法
void init(){
    in=0;out=0;count=0;
  }
} PC;

生产者和消费者逻辑

void producer() {
        item x;
        while (TRUE) {
            produce an item in nextp;
            PC.put(x);
        }
    }

    void consumer(){
        item x;
        while (TRUE) {
            PC.get(x);
            consume the item in nextc;
        ......
        }
    }

    void main() {
        proceducer();
        consumer();
    }

本文作者:程序员潇然 疯狂的字节X https://crazybytex.com/

管程的解决思路就是将同步的问题封装在管程内部，管程会帮你解决所有的问题

哲学家进餐

由Dijkstra提出并解决的哲学家进餐问题(The Dinning Philosophers Problem)是典型的同步问题。 该问题是描述有五个哲学家共用一张圆桌，分别坐在周围的五张椅子上，在圆桌上有五个碗和五只筷子，他们的生活方式是交替地进行思考和进餐。 平时，一个哲学家进行思考，饥饿时便试图取用其左右最靠近他的筷子，只有在他拿到两只筷子时才能进餐。 进餐完毕，放下筷子继续思考。 灰色大圆桌，黄色凳子，每个人左右各有一根筷子，小圆点表示碗。（尽管画的像乌龟，但这真的是桌子 ￣□￣｜｜）

记录型信号量机制

放在桌子上的筷子是临界资源，同一根筷子不可能被两个人同时使用，所以每一根筷子都是一个共享资源 需要使用五个信号量表示，五个信号量每个表示一根筷子

当哲学家饥饿时，总是先去拿他左边的筷子，即执行wait(chopstick[i])； 成功后，再去拿他右边的筷子，即执行wait(chopstick[(i+1)mod 5])；又成功后便可进餐。（i+1)mod 5 是为了处理第五个人右边的是第一个的问题 ） 进餐完毕，又先放下他左边的筷子，然后再放右边的筷子。

//定义五个信号量
//为简单起见，假定数组起始下标为1
//信号量全部初始化为1
semaphore chopstick[5]={1,1,1,1,1};
do{
  //按照我们上面图中所示，第 i号哲学家，左手边为i号筷子，右手边是 (i+1)%5
  wait(chopstick[i]);//等待左手边的，
  wait(chopstick[(i+1)%5]);]);//等待右手边的
  // 进餐......
  signal(chopstick[i]);//释放左手边的
  signal(chopstick[(i+1)%5])//释放右手边的
  // 思考......
} while(TRUE);

通过这种算法可以保证相邻的两个哲学家之间不会出现问题，但是一旦五个人同时拿起左边的筷子，都等待右边的筷子，将会出现死锁 有几种解决思路 (1)至多只允许有四位哲学家同时去拿左边的筷子 可以保证肯定会空余一根筷子，并且没拿起筷子的这个人的左手边的这一根，肯定是已经拿起左手边筷子的某一个人的右手边，所以肯定不会死锁 (2) 仅当哲学家的左、右两只筷子均可用时，才允许他拿起筷子进餐。 也就是AND机制，将左右操作转化为“原子” (3) 规定奇数号哲学家先拿他左边的筷子，然后再去拿右边的筷子，而偶数号哲学家则相反。 如上图所示，1抢1号筷子，2号和3号哲学家竞争3号筷子，4号和5号哲学家竞争5号筷子，所有人都是先竞争奇数，然后再去竞争偶数

这一条是为了所有的人都会先竞争奇数号筷子，那么也就是最多三个人抢到了奇数号筷子，有两个人第一步奇数号筷子都没抢到的这一轮就相当于出局了 三个人，还有两个偶数号筷子，必然会有一个人抢得到

AND型信号量

哲学家进餐需要左手和右手的筷子，所以可以将左右手筷子的获取操作原子化，借助于AND型信号量

//定义五个信号量
//为简单起见，假定数组起始下标为1
//信号量全部初始化为1
semaphore chopstick[5]={1,1,1,1,1};

do{
  //按照我们上面图中所示，第 i号哲学家，左手边为i号筷子，右手边是 (i+1)%5
  Swait(chopstick[i],chopstick[(i+1)%5]))
  // 进餐......
  Ssignal(chopstick[i],chopstick[(i+1)%5]);
  // 思考......
} while(TRUE);

读者写者问题

一个数据文件或记录，可被多个进程共享，我们把只要求读该文件的进程称为“Reader进程” ，其他进程则称为“Writer 进程” 。 允许多个进程同时读一个共享对象，因为读操作不会使数据文件混乱。 但不允许一个Writer 进程和其他Reader 进程或 Writer 进程同时访问共享对象，因为这种访问将会引起混乱。 所谓“读者—写者问题(Reader-Writer Problem)”是指保证一个 Writer 进程必须与其他进程互斥地访问共享对象的同步问题。 读者—写者问题常被用来测试新同步原语。 很显然，只有多个读者时不冲突

记录型信号量机制

读和写之间是互斥的，所以需要一个信号量用于读写互斥Wmutex 另外如果有读的进程存在，另外的进程如果想要读的话，不需要同步也就是Wait(Wmutex)操作； 如果当前没有进程在读，那么需要Wait(Wmutex)操作，所以设置一个变量记录写者个数Readcount，可以用来判断是否需要同步 另外Readcount 会被多个读者进程访问，所以也是临界资源，所以设置一个rmutex 用于互斥访问Readcount

//两个信号量，一个用于读者互斥 readcount ，一个用于读写互斥
semaphore rmutex=l,wmutex=1;
int readcount=0;//初始时读者个数为0
//读者
void reader() {
do{
      wait(rmutex);//读者先获取 readcount
      if(readcount==0){//如果一个读者没有，第一个读者需要与写者互斥访问
      wait(wmutex);
}
readcount++;//读者个数+1
signal(rmutex);//读者个数+1后，可以释放readcount的锁，其他读者可以进来
//开始慢慢读书......
wait(rmutex);//读者结束时，需要获取readcount的锁
readcount--;//退出一个读者
if (readcount==0) {//如果此时一个读者都没有了，还需要释放与读写互斥的锁
    signal(wmutex);
}
signal(rmutex);//释放readcount的锁
  }while(TRUE);
}

void writer(){
do{
  wait(wmutex);//写者必须获得wmutex
  //执行写任务....
  signal(wmutex);//写任务结束后就可以释放锁
}while(TRUE);

}

//主程序
void main() {
  reader();
  writer();
}

写者相对比较简单，获得锁wmutex之后，进行写操作，否则等待wmutex 读者也是需要先获得锁，读操作后释放锁，但是因为多个读者之间互不影响，所以使用readcount记录读者个数，只有第一个读者才需要竞争wmutex，只有最后一个读者才需要释放wmutex readcount作为读者之间的竞争资源，所以对readcount进行操作的时候也需要进行加锁

信号量集机制

将读者写者的问题复杂化一点，它增加了一个限制，即最多只允许 N个读者同时读。 在上面的解决方法中，可以不使用rmutex控制对readcount的互斥，可以构造一个读者个数的信号量readcountmutex，初始值设置为N 每次新增一个读者时，wait(readcountmutex)，一个读者离开时signal(readcountmutex)

也可以使用信号量集机制

int N;//最大的读者个数，也就是相当于图书馆的空位子，初始时空位子为N

semaphore L=N, mx=1;//定义两个信号量资源L和mx，分别用于控制读者个数限制和读写（写写)

void reader() {

do{

    Swait(L, 1, 1);//获取空位子L，每次获取1个，>=1时可分配

    Swait(mx, 1, 0);//获取与写的互斥量mx，每次获取0个，>=1时可分配，如果mx为1，也就是没有写者，读者都可以进来，否则一个都进不来

    //进行一些读操作

    Ssignal(L, 1);//释放一个单位的资源L

  }while(TRUE);

}


void writer() {

do{

  Swait(mx,1,1; L,N,0);//获得资源mx，每次获取1个，>=1时分配，获得资源L，每次获得0个，>=N时即可分配

  //进行一些写操作

  Ssignal(mx, 1);//释放资源mx

  }while(TRUE);

}


void main(){

  reader();

  writer();

}

Swait(L, 1, 1);用于获取读者空位子没什么好说的 Swait(mx, 1, 0);作为开关，只要mx满足条件>=1，那么就可以无限制的进入（此例中有L的限制），一旦条件不满足，则全都不能进入，满足多读者，有写不能读的情况 对于写者中的Swait(mx,1,1; L,N,0); 他会获取mx，>=1时，获取一个资源，并且当L>=N时，分配0个L资源，也就是说一个读者都没有的时候才行 Swait(mx, 1, 0); 与Swait( L,N,0);都是需求0个，相当于开关判断

总结

以上为借助“进程同步的API”，信号量，管程等方式完成进程同步的经典示例，例子来源于《计算机操作系统》 说白了，就是用 wait(S) Swait(S) signal(S) Ssignal(S)等这些“方法”描述进程同步算法

可能会觉得这些内容乱七八糟的，根本没办法使用，的确这些内容全都没办法直接转变为代码写到你的项目中 但是，这些都是解决问题的思路 不管是信号量还是管程还是什么，不会需要你从头开始实现一个信号量，然后.......也不需要你从头开始实现一个管程，然后...... 不管是操作系统层面，还是编程语言层面，还是具体的API，万变不离其宗

尽管这些wait和signal的确不存在，但是，但是，但是编程语言中很可能已经提供了语意相同的方法供你调用了 也就是说，你只需要理解同步的思路即可，尽管没有我们此处说的wait（S）,但是肯定有对应物。

转载务必注明出处：程序员潇然，疯狂的字节X，https://crazybytex.com/thread-58-1-1.html