进程间通信笔记（5）—互斥锁和条件变量

秒速五厘米 2022-09-25 16:17 271阅读 0赞

## 1.简介 ##

互斥与等待用于线程之间的同步，这将允许多线程或多进程之间共享数据。

生产者消费者模型是一个典型的例子，考虑两种情况：

> 1.生产者的生产的速度快于消费者。  
> 2.消费者的消费的速度快于生产者。

这样通信双方有一方必然阻塞，比如说，上述生产者生产速度快于消费者消费速度的情况，如果使用管道来处理这个问题，当管道被塞满后，内核就在生产者调用`write`时把它投入睡眠，直到管道中有空余的空间。这就是同步，不过内核帮咱们把这些事情做了，这种类型的同步是隐式的（implicit），类似的使用消息队列的IPC形式，内核依旧会处理同步。

然而，当我们使用共享内存这种IPC形式的时候，就必须执行某种类型的显示（explicit）同步。

互斥锁和条件变量能够帮我们实现显示的同步：

> 1.互斥锁保护持有锁的线程进入临界区后安全的修改线程共享的内容。  
> 2.条件变量可用于等待条件的发生。

使用的互斥锁和条件变量的时候就是注意不要死锁和[虚假唤醒][Link 1]即可。

--------------------

## 2.示例 ##

结合条件变量和互斥锁的应用，可以参考[封装条件变量][Link 2]，介绍了《Linux多线程服务端编程》中一个简单的BlockingQueue（消费>生产的情况）。

因为生产者消费者同时启动线程，这里用到了互斥和等待两种同步。  
代码见`~/unpv22e/mutex/prodcons6.c`

/* include globals */
    #include    "unpipc.h"
    
    #define MAXNITEMS       1000000
    #define MAXNTHREADS         100
    
    
    int     nitems;
    int     buff[MAXNITEMS];
    struct {
      pthread_mutex_t   mutex;
      int               nput;   /* next index to store */
      int               nval;   /* next value to store */
    } put = { PTHREAD_MUTEX_INITIALIZER };//花括号初始化结构体。
    //用于同步buffer的序号和值
    
    struct {
      pthread_mutex_t   mutex;
      pthread_cond_t    cond;
      int               nready; /* number ready for consumer */
    } nready = { PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER };
    //用于同步计数
    
    void    *produce(void *), *consume(void *);//线程声明
    
    /* include main */
    int
    main(int argc, char **argv)
    {
        int         i, nthreads, count[MAXNTHREADS];
        pthread_t   tid_produce[MAXNTHREADS], tid_consume;
    
        if (argc != 3)
            err_quit("usage: prodcons6 <#items> <#threads>");
        //多个生产者，1个消费者
        nitems = min(atoi(argv[1]), MAXNITEMS);
        nthreads = min(atoi(argv[2]), MAXNTHREADS);
    
        Set_concurrency(nthreads + 1);
            /* 4create all producers and one consumer */
        for (i = 0; i < nthreads; i++) {
            count[i] = 0;//count统计每个生产者的“产量”
            Pthread_create(&tid_produce[i], NULL, produce, &count[i]);
        }
        Pthread_create(&tid_consume, NULL, consume, NULL);//生产者
    
            /* wait for all producers and the consumer */
            //线程回收
        for (i = 0; i < nthreads; i++) {
            Pthread_join(tid_produce[i], NULL);
            printf("count[%d] = %d\n", i, count[i]);    
        }
        //线程回收
        Pthread_join(tid_consume, NULL);
    
        exit(0);
    }
    /* end main */
    
    /* include prodcons */
    void *
    produce(void *arg)
    {
        for ( ; ; ) {
            Pthread_mutex_lock(&put.mutex);
            if (put.nput >= nitems) {
                Pthread_mutex_unlock(&put.mutex);
                return(NULL);       /* array is full, we're done */
            }
            buff[put.nput] = put.nval;
            put.nput++;
            put.nval++;
            Pthread_mutex_unlock(&put.mutex);
    
            Pthread_mutex_lock(&nready.mutex);
            if (nready.nready == 0)
                Pthread_cond_signal(&nready.cond);//通知等待线程。
            nready.nready++;
            Pthread_mutex_unlock(&nready.mutex);
    
            *((int *) arg) += 1;
        }
    }
    
    void *
    consume(void *arg)
    {
        int     i;
    
        for (i = 0; i < nitems; i++) {
            Pthread_mutex_lock(&nready.mutex);
            while (nready.nready == 0)//防止虚假唤醒
                Pthread_cond_wait(&nready.cond, &nready.mutex);
            nready.nready--;
            Pthread_mutex_unlock(&nready.mutex);
    
    //这里只打印出错
    //没有同步的多线程会产生混乱
            if (buff[i] != i)
                printf("buff[%d] = %d\n", i, buff[i]);
        }
        return(NULL);
    }
    /* end prodcons */

书中提到一种修改，避免上锁冲突，可以将唤醒等待线程的操作放到临界区外面。

int dosignal;
    lock(&mutex);//临界区
    dosignal=(ready==0);
    ready++;
    unlock(&mutex);
    
    if (dosignal)
        con_signal(&cond);//可以移除临界区。

[Link 1]: http://blog.csdn.net/zhangxiao93/article/details/51801542?locationNum=1
[Link 2]: http://blog.csdn.net/zhangxiao93/article/details/52027302?locationNum=1