C线程,线程池,信号量,锁以及使用条件变量相关

淩亂°似流年 2022-12-11 02:20 180阅读 0赞

### 文章目录 ###

*  线程
 *  线程池
 *  互斥锁
 *  使用条件变量+互斥锁实现生产者消费者
 *  使用信号量实现生产者消费者

# 线程 #

#include <pthread.h>
    #include <stdio.h>
    #include <unistd.h>
    #include <stdlib.h>
    #include <string.h>
    
    /**
     * 线程,linux下也称之为轻量级进程
     * 在内核层面看来,进程和线程是一样的，但线程PCB独立,三级页表(虚拟地址,三级为:页面-页表-页目录)相同
     * 线程可看做是寄存器和栈(用户栈,内核栈)的集合
     *
     * 多个线程也会和其他进程争抢CPU时钟.
     * 当进程创建线程后,自身也会成为线程.
     *
     * 进程分配资源的
     * 线程执行工作的
     * 进程可看做拥有一个线程的进程.
     *
     * 线程引入信号机制后,谁抢到谁处理,可通过设置线程屏蔽字指定谁处理,未决信号集共享
     *
     * @return
     */
    
    //共享全局变量
    int var = 100;
    
    void *run(void *arg) {
        
        int i;
    //
        var = 200;
        i = (int *) arg;
        if (i == 2) {
        
    //        exit(1);//退出进程
    //        return NULL;   //返回调用者
            //退出当前线程,返回的指针必须全局或者malloc分配的,
            // 不能再线程函数的栈上分配,因为其他线程获得指针后,线程已经退出.
            pthread_exit(NULL);
        }
        printf("pid=%d tid=%lu %d  var=%d \n", getpid(), pthread_self(), i, var);
    
        /*
        i = *((int *) arg);
       sleep(i);
       printf("pid=%d tid=%lu %d var=%d \n", getpid(), pthread_self(), i, var);
        */
    
    }
    
    void demo() {
        
        printf("pid=%d tid=%lu \n", getpid(), pthread_self());
        pthread_t tid;
        int ret, i;
        for (i = 0; i < 5; i++) {
        
    
    //        ret = pthread_create(&tid, NULL, run, (void *) &i);     //地址传递,会共享主线程变量值
            ret = pthread_create(&tid, NULL, run, (void *) i);   //值传递
            if (ret != 0) {
        
                perror("error");
            }
        }
        printf("var=%d \n", var);
        //防止主线程执行结束后,清空子线程数据
    //    sleep(i);
    
    //退出当前线程,不会影响到子线程
        pthread_exit(NULL);
    }
    
    struct thre {
        
        int var;
        char str[256];
    };
    
    void *fun(void *args) {
        
        //利用形参,返回
        struct thre *t = (struct thre *) args;
        t = malloc(sizeof(t));
        t->var = 100;
        strcpy(t->str, "hello");
    
        //返回一个动态分配的指针
    /*   struct thre *t;
        t = malloc(sizeof(t));
        t->var = 100;
        strcpy(t->str, "hello");*/
    
    //返回一个非动态分配的指针.可以发现无法接收,因为局部地址,线程结束后,就回收了
        /*   struct thre t;
           t.var = 10;
           strcpy(t.str, "hello");
           printf("var=%d str=%s \n", t.var, t.str);
               return (void *) &t;
           */
    
        //返回一个整型
    //    int t = 10;
        return (void *) t;
    }
    
    void demo2() {
        
        pthread_t *tid;
        pthread_t arg;
        int ret = pthread_create(&tid, NULL, fun, (void *) &arg);
        if (ret != 0) {
        
            perror("error");
            exit(1);
        }
        struct thre t2;
        struct thre *t = &t2;
        //类似waitpid,阻塞等待线程退出,回收线程
        pthread_join(tid, (void **) &t);
        printf("var=%d str=%s \n", t->var, t->str);
    
    
        //接收一个整型
    /*    int t2;
        int *t = &t2;
        pthread_join(tid, (void **) &t);
        printf("var=%d  \n", t);
     */
    
        pthread_exit(NULL);
    }
    
    void *fun3(void *args) {
        
        while (1) {
        
            /* printf("pid=%d tid=%lu \n", getpid(), pthread_self());
             sleep(1);*/
            //添加取消点
            pthread_testcancel();
        }
        return (void *) 66;
    }
    
    void demo3() {
        
        pthread_t pthread;
        int ret = pthread_create(&pthread, NULL, fun3, NULL);
        if (ret != 0) {
        
            perror("error");
            exit(1);
        }
        printf("pid=%d tid=%lu \n", getpid(), pthread_self());
        sleep(5);
        /**终止线程
         *函数若没进入内核,则会不终止.sleep()之所以有效,是因为进入了内核.
         *可添加取消点,会进入内核检测是否取消.
         */
        ret = pthread_cancel(pthread);
        void *tret = NULL;
        pthread_join(pthread, &tret);
        printf("%d \n", tret);
        if (ret != 0) {
        
            perror("error");
            exit(1);
        }
        while (1);
    }
    
    void *fun4(void *args) {
        
    //    while (1);
        return (void *) 66;
    }
    
    void demo4() {
        
        pthread_t pthread;
        int ret = pthread_create(&pthread, NULL, fun4, NULL);
        if (ret != 0) {
        
            fprintf(stderr, "error: %s \n", strerror(ret));
            exit(1);
        }
        /**
         *线程分离
         * 线程终止会自动清理。无需回收
         */
        ret = pthread_detach(pthread);
        if (ret != 0) {
        
            fprintf(stderr, "error: %s \n", strerror(ret));
            exit(1);
        }
        sleep(1);
        //分离后,join会抛出参数无效,这是因为线程分离后,线程直接执行完回收了.
        ret = pthread_join(pthread, NULL);
        if (ret != 0) {
        
            fprintf(stderr, "error: %s \n", strerror(ret));
            exit(1);
        }
        pthread_exit((void *) 0);
    }
    
    void demo5() {
        
        int ret;
        //初始化属性
        pthread_attr_t pa;
        ret = pthread_attr_init(&pa);
        if (ret != 0) {
        
            fprintf(stderr, "error: %s \n", strerror(ret));
            exit(1);
        }
    
        //设置分离属性
        ret = pthread_attr_setdetachstate(&pa, PTHREAD_CREATE_DETACHED);
        if (ret != 0) {
        
            fprintf(stderr, "error: %s \n", strerror(ret));
            exit(1);
        }
    
        pthread_t pthread;
        ret = pthread_create(&pthread, &pa, fun4, NULL);
        if (ret != 0) {
        
            fprintf(stderr, "error: %s \n", strerror(ret));
            exit(1);
        }
    
        //销毁属性
        ret = pthread_attr_destroy(&pa);
        if (ret != 0) {
        
            fprintf(stderr, "error: %s \n", strerror(ret));
            exit(1);
        }
    
        ret = pthread_join(pthread, NULL);
        if (ret != 0) {
        
            fprintf(stderr, "error: %s \n", strerror(ret));
            exit(1);
        }
    
    
        printf("pid=%d tid=%lu  \n", getpid(), pthread_self());
        pthread_exit((void *) 0);
    }
    
    int main() {
        
        //创建线程
    //    demo();
    
    //测试pthread_join
    //    demo2();
    
    //cancel取消线程
    //    demo3();
    
    //线程分离
    //    demo4();
    
        //通过设置线程属性分离
    //    demo5();
    
    
    
        return 0;
    }

# 线程池 #

#include <stdlib.h>
    #include <pthread.h>
    #include <unistd.h>
    #include <stdio.h>
    #include <string.h>
    #include <signal.h>
    #include <errno.h>
    
    #define DEFAULT_TIME 10                 /*10s检测一次*/
    #define MIN_WAIT_TASK_NUM 10            /*如果queue_size > MIN_WAIT_TASK_NUM 添加新的线程到线程池*/
    #define DEFAULT_THREAD_VARY 10          /*每次创建和销毁线程的个数*/
    #define true 1
    #define false 0
    
    typedef struct {
        
        void *(*function)(void *);          /* 函数指针，回调函数 */
        void *arg;                          /* 上面函数的参数 */
    } threadpool_task_t;                    /* 各子线程任务结构体 */
    
    /* 描述线程池相关信息 */
    typedef struct threadpool_t {
        
        pthread_mutex_t lock;               /* 用于锁住本结构体 */
        pthread_mutex_t thread_counter;     /* 记录忙状态线程个数de琐 -- busy_thr_num */
        pthread_cond_t queue_not_full;      /* 当任务队列满时，添加任务的线程阻塞，等待此条件变量 */
        pthread_cond_t queue_not_empty;     /* 任务队列里不为空时，通知等待任务的线程 */
    
        pthread_t *threads;                 /* 存放线程池中每个线程的tid。数组 */
        pthread_t adjust_tid;               /* 存管理线程tid */
        threadpool_task_t *task_queue;      /* 任务队列 */
    
        int min_thr_num;                    /* 线程池最小线程数 */
        int max_thr_num;                    /* 线程池最大线程数 */
        int live_thr_num;                   /* 当前存活线程个数 */
        int busy_thr_num;                   /* 忙状态线程个数 */
        int wait_exit_thr_num;              /* 要销毁的线程个数 */
    
        int queue_front;                    /* task_queue队头下标 */
        int queue_rear;                     /* task_queue队尾下标 */
        int queue_size;                     /* task_queue队中实际任务数 */
        int queue_max_size;                 /* task_queue队列可容纳任务数上限 */
    
        int shutdown;                       /* 标志位，线程池使用状态，true或false */
    }threadpool_t ;
    
    /**
     * @function void *threadpool_thread(void *threadpool)
     * @desc the worker thread
     * @param threadpool the pool which own the thread
     */
    void *threadpool_thread(void *threadpool);
    
    /**
     * @function void *adjust_thread(void *threadpool);
     * @desc manager thread
     * @param threadpool the threadpool
     */
    void *adjust_thread(void *threadpool);
    
    /**
     * check a thread is alive
     */
    int is_thread_alive(pthread_t tid);
    int threadpool_free(threadpool_t *pool);
    
    threadpool_t *threadpool_create(int min_thr_num, int max_thr_num, int queue_max_size)
    {
        
        int i;
        threadpool_t *pool = NULL;
        do {
        
            if((pool = (threadpool_t *)malloc(sizeof(threadpool_t))) == NULL) {
        
                printf("malloc threadpool fail");
                break;/*跳出do while*/
            }
    
            pool->min_thr_num = min_thr_num;
            pool->max_thr_num = max_thr_num;
            pool->busy_thr_num = 0;
            pool->live_thr_num = min_thr_num;               /* 活着的线程数 初值=最小线程数 */
            pool->queue_size = 0;                           /* 有0个产品 */
            pool->queue_max_size = queue_max_size;
            pool->queue_front = 0;
            pool->queue_rear = 0;
            pool->shutdown = false;                         /* 不关闭线程池 */
    
            /* 根据最大线程上限数， 给工作线程数组开辟空间, 并清零 */
            pool->threads = (pthread_t *)malloc(sizeof(pthread_t)*max_thr_num);
            if (pool->threads == NULL) {
        
                printf("malloc threads fail");
                break;
            }
            memset(pool->threads, 0, sizeof(pthread_t)*max_thr_num);
    
            /* 队列开辟空间 */
            pool->task_queue = (threadpool_task_t *)malloc(sizeof(threadpool_task_t)*queue_max_size);
            if (pool->task_queue == NULL) {
        
                printf("malloc task_queue fail");
                break;
            }
    
            /* 初始化互斥琐、条件变量 */
            if (pthread_mutex_init(&(pool->lock), NULL) != 0
                || pthread_mutex_init(&(pool->thread_counter), NULL) != 0
                || pthread_cond_init(&(pool->queue_not_empty), NULL) != 0
                || pthread_cond_init(&(pool->queue_not_full), NULL) != 0)
            {
        
                printf("init the lock or cond fail");
                break;
            }
    
            /* 启动 min_thr_num 个 work thread */
            for (i = 0; i < min_thr_num; i++) {
        
                pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void *)pool);/*pool指向当前线程池*/
                printf("start thread 0x%x...\n", (unsigned int)pool->threads[i]);
            }
            pthread_create(&(pool->adjust_tid), NULL, adjust_thread, (void *)pool);/* 启动管理者线程 */
    
            return pool;
    
        } while (0);
    
        threadpool_free(pool);      /* 前面代码调用失败时，释放poll存储空间 */
    
        return NULL;
    }
    
    /* 向线程池中 添加一个任务 */
    int threadpool_add(threadpool_t *pool, void*(*function)(void *arg), void *arg)
    {
        
        pthread_mutex_lock(&(pool->lock));
    
        /* ==为真，队列已经满， 调wait阻塞 */
        while ((pool->queue_size == pool->queue_max_size) && (!pool->shutdown)) {
        
            pthread_cond_wait(&(pool->queue_not_full), &(pool->lock));
        }
        if (pool->shutdown) {
        
            pthread_mutex_unlock(&(pool->lock));
        }
    
        /* 清空 工作线程 调用的回调函数 的参数arg */
        if (pool->task_queue[pool->queue_rear].arg != NULL) {
        
            free(pool->task_queue[pool->queue_rear].arg);
            pool->task_queue[pool->queue_rear].arg = NULL;
        }
        /*添加任务到任务队列里*/
        pool->task_queue[pool->queue_rear].function = function;
        pool->task_queue[pool->queue_rear].arg = arg;
        pool->queue_rear = (pool->queue_rear + 1) % pool->queue_max_size;       /* 队尾指针移动, 模拟环形 */
        pool->queue_size++;
    
        /*添加完任务后，队列不为空，唤醒线程池中 等待处理任务的线程*/
        pthread_cond_signal(&(pool->queue_not_empty));
        pthread_mutex_unlock(&(pool->lock));
    
        return 0;
    }
    
    /* 线程池中各个工作线程 */
    void *threadpool_thread(void *threadpool)
    {
        
        threadpool_t *pool = (threadpool_t *)threadpool;
        threadpool_task_t task;
    
        while (true) {
        
            /* Lock must be taken to wait on conditional variable */
            /*刚创建出线程，等待任务队列里有任务，否则阻塞等待任务队列里有任务后再唤醒接收任务*/
            pthread_mutex_lock(&(pool->lock));
    
            /*queue_size == 0 说明没有任务，调 wait 阻塞在条件变量上, 若有任务，跳过该while*/
            while ((pool->queue_size == 0) && (!pool->shutdown)) {
        
                printf("thread 0x%x is waiting\n", (unsigned int)pthread_self());
                pthread_cond_wait(&(pool->queue_not_empty), &(pool->lock));
    
                /*清除指定数目的空闲线程，如果要结束的线程个数大于0，结束线程*/
                if (pool->wait_exit_thr_num > 0) {
        
                    pool->wait_exit_thr_num--;
    
                    /*如果线程池里线程个数大于最小值时可以结束当前线程*/
                    if (pool->live_thr_num > pool->min_thr_num) {
        
                        printf("thread 0x%x is exiting\n", (unsigned int)pthread_self());
                        pool->live_thr_num--;
                        pthread_mutex_unlock(&(pool->lock));
                        pthread_exit(NULL);
                    }
                }
            }
    
            /*如果指定了true，要关闭线程池里的每个线程，自行退出处理*/
            if (pool->shutdown) {
        
                pthread_mutex_unlock(&(pool->lock));
                printf("thread 0x%x is exiting\n", (unsigned int)pthread_self());
                pthread_exit(NULL);     /* 线程自行结束 */
            }
    
            /*从任务队列里获取任务, 是一个出队操作*/
            task.function = pool->task_queue[pool->queue_front].function;
            task.arg = pool->task_queue[pool->queue_front].arg;
    
            pool->queue_front = (pool->queue_front + 1) % pool->queue_max_size;       /* 出队，模拟环形队列 */
            pool->queue_size--;
    
            /*通知可以有新的任务添加进来*/
            pthread_cond_broadcast(&(pool->queue_not_full));
    
            /*任务取出后，立即将 线程池琐 释放*/
            pthread_mutex_unlock(&(pool->lock));
    
            /*执行任务*/
            printf("thread 0x%x start working\n", (unsigned int)pthread_self());
            pthread_mutex_lock(&(pool->thread_counter));                            /*忙状态线程数变量琐*/
            pool->busy_thr_num++;                                                   /*忙状态线程数+1*/
            pthread_mutex_unlock(&(pool->thread_counter));
            (*(task.function))(task.arg);                                           /*执行回调函数任务*/
            //task.function(task.arg);                                              /*执行回调函数任务*/
    
            /*任务结束处理*/
            printf("thread 0x%x end working\n", (unsigned int)pthread_self());
            pthread_mutex_lock(&(pool->thread_counter));
            pool->busy_thr_num--;                                       /*处理掉一个任务，忙状态数线程数-1*/
            pthread_mutex_unlock(&(pool->thread_counter));
        }
    
        pthread_exit(NULL);
    }
    
    /* 管理线程 */
    void *adjust_thread(void *threadpool)
    {
        
        int i;
        threadpool_t *pool = (threadpool_t *)threadpool;
        while (!pool->shutdown) {
        
    
            sleep(DEFAULT_TIME);                                    /*定时 对线程池管理*/
    
            pthread_mutex_lock(&(pool->lock));
            int queue_size = pool->queue_size;                      /* 关注 任务数 */
            int live_thr_num = pool->live_thr_num;                  /* 存活 线程数 */
            pthread_mutex_unlock(&(pool->lock));
    
            pthread_mutex_lock(&(pool->thread_counter));
            int busy_thr_num = pool->busy_thr_num;                  /* 忙着的线程数 */
            pthread_mutex_unlock(&(pool->thread_counter));
    
            /* 创建新线程 算法： 任务数大于最小线程池个数, 且存活的线程数少于最大线程个数时 如：30>=10 && 40<100*/
            if (queue_size >= MIN_WAIT_TASK_NUM && live_thr_num < pool->max_thr_num) {
        
                pthread_mutex_lock(&(pool->lock));
                int add = 0;
    
                /*一次增加 DEFAULT_THREAD 个线程*/
                for (i = 0; i < pool->max_thr_num && add < DEFAULT_THREAD_VARY
                            && pool->live_thr_num < pool->max_thr_num; i++) {
        
                    if (pool->threads[i] == 0 || !is_thread_alive(pool->threads[i])) {
        
                        pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void *)pool);
                        add++;
                        pool->live_thr_num++;
                    }
                }
    
                pthread_mutex_unlock(&(pool->lock));
            }
    
            /* 销毁多余的空闲线程 算法：忙线程X2 小于 存活的线程数 且 存活的线程数 大于 最小线程数时*/
            if ((busy_thr_num * 2) < live_thr_num  &&  live_thr_num > pool->min_thr_num) {
        
    
                /* 一次销毁DEFAULT_THREAD个线程, 隨機10個即可 */
                pthread_mutex_lock(&(pool->lock));
                pool->wait_exit_thr_num = DEFAULT_THREAD_VARY;      /* 要销毁的线程数 设置为10 */
                pthread_mutex_unlock(&(pool->lock));
    
                for (i = 0; i < DEFAULT_THREAD_VARY; i++) {
        
                    /* 通知处在空闲状态的线程, 他们会自行终止*/
                    pthread_cond_signal(&(pool->queue_not_empty));
                }
            }
        }
    
        return NULL;
    }
    
    int threadpool_destroy(threadpool_t *pool)
    {
        
        int i;
        if (pool == NULL) {
        
            return -1;
        }
        pool->shutdown = true;
    
        /*先销毁管理线程*/
        pthread_join(pool->adjust_tid, NULL);
    
        for (i = 0; i < pool->live_thr_num; i++) {
        
            /*通知所有的空闲线程*/
            pthread_cond_broadcast(&(pool->queue_not_empty));
        }
        for (i = 0; i < pool->live_thr_num; i++) {
        
            pthread_join(pool->threads[i], NULL);
        }
        threadpool_free(pool);
    
        return 0;
    }
    
    int threadpool_free(threadpool_t *pool)
    {
        
        if (pool == NULL) {
        
            return -1;
        }
    
        if (pool->task_queue) {
        
            free(pool->task_queue);
        }
        if (pool->threads) {
        
            free(pool->threads);
            pthread_mutex_lock(&(pool->lock));
            pthread_mutex_destroy(&(pool->lock));
            pthread_mutex_lock(&(pool->thread_counter));
            pthread_mutex_destroy(&(pool->thread_counter));
            pthread_cond_destroy(&(pool->queue_not_empty));
            pthread_cond_destroy(&(pool->queue_not_full));
        }
        free(pool);
        pool = NULL;
    
        return 0;
    }
    
    int threadpool_all_threadnum(threadpool_t *pool)
    {
        
        int all_threadnum = -1;
        pthread_mutex_lock(&(pool->lock));
        all_threadnum = pool->live_thr_num;
        pthread_mutex_unlock(&(pool->lock));
        return all_threadnum;
    }
    
    int threadpool_busy_threadnum(threadpool_t *pool)
    {
        
        int busy_threadnum = -1;
        pthread_mutex_lock(&(pool->thread_counter));
        busy_threadnum = pool->busy_thr_num;
        pthread_mutex_unlock(&(pool->thread_counter));
        return busy_threadnum;
    }
    
    int is_thread_alive(pthread_t tid)
    {
        
        int kill_rc = pthread_kill(tid, 0);     //发0号信号，测试线程是否存活
        if (kill_rc == ESRCH) {
        
            return false;
        }
    
        return true;
    }
    
    /*测试*/
    
    #if 1
    /* 线程池中的线程，模拟处理业务 */
    void *process(void *arg)
    {
        
        printf("thread 0x%x working on task %d\n ",(unsigned int)pthread_self(),*(int *)arg);
        sleep(1);
        printf("task %d is end\n",*(int *)arg);
    
        return NULL;
    }
    int main(void)
    {
        
        /*threadpool_t *threadpool_create(int min_thr_num, int max_thr_num, int queue_max_size);*/
    
        threadpool_t *thp = threadpool_create(3,100,100);/*创建线程池，池里最小3个线程，最大100，队列最大100*/
        printf("pool inited");
    
        //int *num = (int *)malloc(sizeof(int)*20);
        int num[20], i;
        for (i = 0; i < 20; i++) {
        
            num[i]=i;
            printf("add task %d\n",i);
            threadpool_add(thp, process, (void*)&num[i]);     /* 向线程池中添加任务 */
        }
        sleep(10);                                          /* 等子线程完成任务 */
        threadpool_destroy(thp);
    
        return 0;
    }
    
    #endif

# 互斥锁 #

#include <pthread.h>
    #include <stdio.h>
    #include <unistd.h>
    #include <stdlib.h>
    #include <string.h>
    
    /**
     * 线程同步
     * @return
     */
    //定义互斥锁
    pthread_mutex_t mutex;
    //静态初始化
    //pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;
    
    void *fun(void *arg) {
        
        srand(time(NULL));
    
        while (1) {
        
            //加锁,失败阻塞
            pthread_mutex_lock(&mutex);
    
    
            printf(" hello ");
            sleep(rand() % 1);
            printf(" world \n");
    
    
            sleep(rand() % 1);
    
            //解锁
            pthread_mutex_unlock(&mutex);
        }
        return NULL;
    }
    
    void lock() {
        
        srand(time(NULL));
    
        //动态初始化锁
        int ret = pthread_mutex_init(&mutex, NULL);
        if (ret != 0) {
        
            fprintf(stderr, "error: %s \n", strerror(ret));
            exit(1);
        }
    
    
        pthread_t tid;
        pthread_create(&tid, NULL, fun, NULL);
        while (1) {
        
            //加锁
            pthread_mutex_lock(&mutex);
    
            //再次加锁,会造成死锁
            // pthread_mutex_lock(&mutex);
    
            //试图加锁,失败则立即返回
    //        pthread_mutex_trylock(&mutex);
    
            printf(" HELLO ");
            sleep(rand() % 1);
            printf(" WORLD \n");
            sleep(rand() % 1);
    
            //解锁
            pthread_mutex_unlock(&mutex);
        }
        pthread_join(tid, NULL);
    
        //销毁锁
        pthread_mutex_destroy(&mutex);
    
    }
    
    
    int main() {
        
        //独占锁
    //  lock();
    
    /**
     * 读写锁
     *
     *  pthread_rwlock_init();
        pthread_rwlock_wrlock();
        pthread_rwlock_trywrlock();
        pthread_rwlock_rdlock();
        pthread_rwlock_tryrdlock();
        pthread_rwlock_unlock();
        pthread_rwlock_destroy();
     */
    
    
    
    
        return 0;
    }

# 使用条件变量+互斥锁实现生产者消费者 #

#include <pthread.h>
    #include <stdio.h>
    #include <unistd.h>
    #include <stdlib.h>
    #include <string.h>
    
    /**
     * 使用条件变量+互斥锁实现生产者消费者队列
     * @return
     */
    pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
    pthread_cond_t condition = PTHREAD_COND_INITIALIZER;
    struct msg {
        
        struct msg *next;
        int num;
    };
    struct msg *head;
    
    void *consumer(void *p) {
        
        struct msg *m;
        while (1) {
        
            pthread_mutex_lock(&lock);
            /**若修改为if会发生如下情况:
             * AB线程阻塞在这里.
             * C线程生产
             * A线程接收到信号,唤醒.
             * 执行消费.
             * 此时B线程也接收到信号,因为A线程抢到了锁,所以阻塞条件变量变成阻塞等待锁释放上
             * 当A释放锁,则B获取到锁.然后若head没值,则会异常.
             *
             */
            while (head == NULL) {
        
                pthread_cond_wait(&condition, &lock);
            }
            m = head;
            head = m->next;
            pthread_mutex_unlock(&lock);
            printf("consumer id=%lu,val=%d \n",pthread_self(), m->num);
            free(m);
            sleep(rand() % 3);
        }
    }
    
    void *producer(void *p) {
        
        struct msg *m;
        while (1) {
        
            m=malloc(sizeof(struct msg));
            m->num = rand() % 3;
    
            pthread_mutex_lock(&lock);
    
            m->next = head;
            head = m;
            printf("produce val=%d \n", m->num);
            pthread_mutex_unlock(&lock);
    
            //唤醒阻塞在条件变量上的至少一个线程
            pthread_cond_signal(&condition);
            //唤醒阻塞条件变量上的所有线程
    //        pthread_cond_broadcast(&condition);
    
            sleep(rand() % 2);
        }
    }
    
    int main() {
        
        pthread_t pid, cid,cid2,cid3;
    
        pthread_create(&pid, NULL, producer, NULL);
        pthread_create(&cid, NULL, consumer, NULL);
        pthread_create(&cid2, NULL, consumer, NULL);
        pthread_create(&cid3, NULL, consumer, NULL);
    
        pthread_join(pid, NULL);
        pthread_join(cid, NULL);
        pthread_join(cid2, NULL);
        pthread_join(cid3, NULL);
        return 0;
    }

# 使用信号量实现生产者消费者 #

#include <semaphore.h>
    #include <pthread.h>
    #include <stdlib.h>
    #include <stdio.h>
    #include <unistd.h>
    
    /**
     * 信号量
     * @return
     */
    #define  NUM 5
    int queue[NUM];
    sem_t blank_number, product_number;
    
    void *consumer(void *p) {
        
        int i = 0;
        while (1) {
        
            //初次调用,这里一直阻塞,等待生产后释放信号量.相当于独占锁
            sem_wait(&product_number);
            queue[i] = rand() % 1000 + 1;
            printf("consumer %d \n", queue[i]);
            queue[i]=0;
            sem_post(&blank_number);
            i = (i + 1) % NUM;
            sleep( rand() % 3);
        }
    }
    
    void *producer(void *p) {
        
        int i = 0;
        while (1) {
        
            //信号量数量-1,为0则阻塞
            sem_wait(&blank_number);
            queue[i] = rand() % 1000 + 1;
            printf("produce %d \n", queue[i]);
            //信号量+=1,为设定数量则阻塞
    //        sem_post(&product_number);
            i = (i + 1) % NUM;
            sleep( rand() % 1);
        }
    }
    
    int main() {
        
        pthread_t pid, cid;
        /** 信号量
         * 0:线程同步
         * 1:进程同步
         * value:指定同时访问的线程数
         *
         */
        sem_init(&blank_number, 0, NUM);
        sem_init(&product_number, 0, 0);
    
        pthread_create(&pid, NULL, producer, NULL);
        pthread_create(&cid, NULL, consumer, NULL);
    
        pthread_join(pid, NULL);
        pthread_join(cid, NULL);
    
        //销毁信号量
        sem_destroy(&blank_number);
        sem_destroy(&product_number);
    
        return 0;
    }