生产者和消费者模型 和 条件变量

    生产者
    容器装满后不再生产阻塞 并通知消费者去消费
    容器
    消费者
    容器无内容后不再消费阻塞 并通知生产者去生产

    1. /*
    2. 生产者消费者模型 也叫有限缓冲模型
    3. 生产者向容器中添加节点 消费者从容器中删除节点
    5. 条件变量 不是锁  但是可以在某个条件满足后阻塞线程或解阻塞线程
    6. pthread_cond_t 
    7. int pthread_cond_init(pthread_cond_t *restrict cond, pthread_condattr_t *restrict cond_attr);
    8. 初始化条件变量
    9. int pthread_cond_destroy(pthread_cond_t *cond);
    10. 销毁条件变量
    11. int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);
    12. 阻塞函数  等待被唤醒(在信号或广播唤醒前一直等待)
    13. pthread_cond_wait的线程在等待期间会将锁给让出去 当前线程在等待时是不拿锁的
    14. 在被唤醒等待结束不阻塞时 锁又会还给这个在等待的线程
    15. int pthread_cond_timedwait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex, const struct timespec *restrict abstime);
    16. 阻塞函数  等待被唤醒(在信号或广播唤醒或阻塞的实际还没有超过设定的abstime前 一直等待)
    17. 等待期间会将锁给让出去 当前线程在等待时是不拿锁的
    18. 在被唤醒等待结束不阻塞时 锁又会还给这个在等待的线
    19. int pthread_cond_signal(pthread_cond_t *cond);
    20. 唤醒1个或多个在等待的线程
    21. int pthread_cond_broadcast(pthread_cond_t *cond);
    22. 唤醒所有在等待的线程
    23. */
    24. #include <stdio.h>
    25. #include <pthread.h>
    26. #include <string.h>
    27. #include <unistd.h>
    28. #include <stdlib.h>      //malloc
    29. pthread_rwlock_t rwlock; //全局读写锁量
    30. void *produce(void *arg);
    31. void *custom(void *arg);
    32. //使用链表作为容器 假设这个容器不会满 但是会空
    33. typedef struct Node
    34. {
    35.     int data;
    36.     struct Node *next;
    37. } Node;
    38. Node *head = NULL; //头节点
    39. //创建条件变量
    40. pthread_cond_t cond;
    41. int main()
    42. {
    43.     //初始化互斥量
    44.     pthread_rwlock_init(&rwlock, NULL);
    45.     //初始化条件变量
    46.     pthread_cond_init(&cond, NULL);
    47.     //  创建3个子线程 主线程只用于管理系统资源
    48.     pthread_t ptids[5],
    49.         ctids[5]; //线程数组 5个生产者 5个消费者
    50.     for (int i = 0; i < 5; i++)
    51.     {
    52.         int ret = pthread_create(&ptids[i], NULL, produce, NULL);
    53.         if (ret != 0)
    54.         {
    55.             printf("%s\n", strerror(ret));
    56.         }
    57.         //不join了 直接将10个线程都分离 由系统回收
    58.         pthread_detach(ptids[i]);
    59.     }
    60.     for (int i = 0; i < 5; i++)
    61.     {
    62.         int ret = pthread_create(&ctids[i], NULL, custom, NULL);
    63.         if (ret != 0)
    64.         {
    65.             printf("%s\n", strerror(ret));
    66.         }
    67.         //不join了 直接将10个线程都分离 由系统回收
    68.         pthread_detach(ctids[i]);
    69.     }
    70.     while (1)
    71.     {
    72.         sleep(10);
    73.         break;
    75.     }
    76.     pthread_cond_destroy(&cond);
    77.     //子线程执行10秒后就将锁销毁
    78.     pthread_rwlock_destroy(&rwlock);
    79.     //pthread_exit(NULL); //让主线程退出 当主线程退出时 不会影响其他正常运行的线程
    80.     return 0; //不会执行
    81. }
    82. // 这整个链表就像栈 生产和消费的过程就是出栈和入栈 生产出来加在队首 消费也优先消费队首的节点
    83. // 生产和消费都是写操作
    84. void *produce(void *arg) //生产线程 向容器添加内容 不断创建节点添加到链表中--新节点直接作为新的头节点
    85. {
    86.     while (1)
    87.     {
    88.         pthread_rwlock_wrlock(&rwlock);
    89.         Node *new_node = (Node *)malloc(sizeof(Node));
    90.         new_node->data = rand() % 1000;
    91.         new_node->next = head;
    92.         head = new_node;
    93.         printf("%ld produce node %d\n
    94. ", pthread_self(), new_node->data);
    95.         //只要生产了一个就通知消费者消费 唤醒在等待的消费者
    96.         pthread_cond_signal(&cond);
    97.         pthread_rwlock_unlock(&rwlock);
    98.         usleep(1000); //每个线程在产生一个新节点后至少1ms不生产新节点
    99.     }
    100.     return NULL;
    101. }
    102. void *custom(void *arg)
    103. {
    104.     while (1)
    105.     {
    106.         pthread_rwlock_wrlock(&rwlock);
    107.         if (head == NULL)
    108.         {
    109.             //1head此时为null说明生产线程还没有生产节点 容器为空 需要将锁给生产线程，让生产线程先生产节点先
    110.             // pthread_rwlock_unlock(&rwlock);
    111.             // usleep(1000); //每个线程在消费一个节点后至少1ms不争夺锁
    112.             // continue;
    113.             //2 没有数据阻塞等待 当生产者生产了一个节点会发信号通知 这里就会解除阻塞
    114.             pthread_cond_wait(&cond, &rwlock); //当前消费者拿着锁在等待 生产者拿不到锁怎么生产？
    115.             //并不会这样 pthread_cond_wait在等待期间会将锁给让出去 当前线程在等待时是不拿锁的
    116.             //当唤醒后不再等待 锁又归还给当前线程
    117.             pthread_rwlock_unlock(&rwlock);
    118.             continue;
    119.         }
    120.         Node *temp = head;
    121.         head = head->next;
    122.         printf("%ld custom node %d\n
    123. ", pthread_self(), temp->data);
    124.         free(temp); //释放头节点
    125.         pthread_rwlock_unlock(&rwlock);
    126.         usleep(1000); //每个线程在消费一个节点后至少1ms不消费新节点
    127.     }
    128.     return NULL;
    129. }



    信号量
    相比条件变量使用起来更灵活方便

    1. /*
    2. 信号量(信号灯) 与条件变量类似 用于阻塞线程 但一样不是锁
    3. 类型 sem_t
    4. #include <semaphore.h>
    6. int sem_init(sem_t *sem, int pshared, unsigned int value);
    7. 初始化信号量
    8. sem 要初始化的信号量
    9. pshared 此信号量用于线程间还是进程间 为0是线程 非0是进程(信号量在共享内存中)
    10. 但LinuxThreads仅实现线程间共享，因此所有非0值的pshared输入都将使sem_init()返回-1，且置errno为ENOSYS。
    11. value 信号量中断的值(总共有几个灯)
    12. int sem_destroy(sem_t *sem);
    13. 销毁信号量 释放资源
    14. 释放自己占用的一切资源，被注销的信号量sem要求已没有线程在等待该信号量了，否则返回-1，且置errno为EBUSY。
    15. wait对信号量加锁
    16. int sem_wait(sem_t *sem);
    17. 若当前亮着的灯为0个则阻塞直到亮灯不为0个  不为0不会阻塞 将亮着的灯关掉一个(-1)
    18. int sem_trywait(sem_t *sem);
    19. sem_trywait()为sem_wait()的非阻塞版，如果信号量计数大于0，则信号量立即减1并返回0，否则立即返回-1，errno置为EAGAIN。
    20. int sem_timedwait(sem_t *sem, const struct timespec *abs_timeout);
    21. sem_timedwait和sem_wait一样当信号量计数为0会阻塞,但sem_timedwait的abs_timeout参数指定了调用应该阻塞的时间限制
    22. 当阻塞超时时返回失败(errno 设置为 ETIMEDOUT)
    23. 尽量不要使用sem_timedwait函数来实现延时等待的功能，若要使用该延时等待的功能，建议使用sem_trywait+usleep 实现的延时阻塞
    24. int sem_post(sem_t *sem);
    25. 解锁一个信号量(开一个灯(+1))  唤醒在wait的线程(它获得了一个信号量锁)
    26. 它是一个“原子操作”，即同时对同一个信号量做加“1”操作的两个线程是不会冲突的,一个先加1一个后加1最后是加2
    27. int sem_getvalue(sem_t *sem, int *sval);
    28. 读取sem中信号量计数，存于*sval中，并返回0。
    29. */
    30. #include <stdio.h>
    31. #include <pthread.h>
    32. #include <string.h>
    33. #include <unistd.h>
    34. #include <stdlib.h> //malloc
    35. #include <semaphore.h>
    36. pthread_rwlock_t rwlock; //全局读写锁量
    37. void *produce(void *arg);
    38. void *custom(void *arg);
    39. //使用链表作为容器 假设这个容器不会满 但是会空
    40. typedef struct Node
    41. {
    42.     int data;
    43.     struct Node *next;
    44. } Node;
    45. Node *head = NULL; //头节点
    46. //创建条两个信号量
    47. sem_t csem, psem;
    48. int main()
    49. {
    50.     //初始化互斥量
    51.     pthread_rwlock_init(&rwlock, NULL);
    52.     //初始化信号量
    53.     //这个10限定了容器中最多只会有10个节点 生产者灭的灯数+消费者亮的灯数=消费者灭的灯数+生产者亮的灯数=10 这个式子永远是恒等的
    54.     sem_init(&psem, 0, 10); //生产者的信号量初始值为10  这个值其实可以设置为容器的上限
    55.     sem_init(&csem, 0, 0);  //消费的信号量初始值为0 这意味着消费者一开始是阻塞的(因为生产者还没生产 容器为空不允许消费)
    56.     //  创建子线程 主线程只用于管理系统资源
    57.     pthread_t ptids[5],
    58.         ctids[5]; //线程数组 5个生产者 5个消费者
    59.     for (int i = 0; i < 5; i++)
    60.     {
    61.         int ret = pthread_create(&ptids[i], NULL, produce, NULL);
    62.         if (ret != 0)
    63.         {
    64.             printf("%s\n", strerror(ret));
    65.         }
    66.         //不join了 直接将10个线程都分离 由系统回收
    67.         pthread_detach(ptids[i]);
    68.     }
    69.     for (int i = 0; i < 5; i++)
    70.     {
    71.         int ret = pthread_create(&ctids[i], NULL, custom, NULL);
    72.         if (ret != 0)
    73.         {
    74.             printf("%s\n", strerror(ret));
    75.         }
    76.         //不join了 直接将10个线程都分离 由系统回收
    77.         pthread_detach(ctids[i]);
    78.     }
    79.     while (1)
    80.     {
    81.         sleep(10);
    82.         break;
    83.     }
    84.     sem_destroy(&csem);
    85.     sem_destroy(&psem);
    86.     //子线程执行10秒后就将锁销毁
    87.     pthread_rwlock_destroy(&rwlock);
    88.     // pthread_exit(NULL); //让主线程退出 当主线程退出时 不会影响其他正常运行的线程
    89.     return 0; //不会执行
    90. }
    91. // 这整个链表就像栈 生产和消费的过程就是出栈和入栈 生产出来加在队首 消费也优先消费队首的节点
    92. // 生产和消费都是写操作
    93. //生产者灭的灯数+消费者亮的灯数=消费者灭的灯数+生产者亮的灯数=10 这个式子永远是恒等的
    94. void *produce(void *arg) //生产线程 向容器添加内容 不断创建节点添加到链表中--新节点直接作为新的头节点
    95. {
    96.     while (1)
    97.     {
    98.         sem_wait(&psem); //每生产一次 生产者亮灯数减1  当亮灯数为0 阻塞等待消费者消费
    99.         pthread_rwlock_wrlock(&rwlock);
    100.         Node *new_node = (Node *)malloc(sizeof(Node));
    101.         new_node->data = rand() % 1000;
    102.         new_node->next = head;
    103.         head = new_node;
    104.         printf("%ld produce node %d\n", pthread_self(), new_node->data);
    105.         //只要生产了一个就通知消费者消费 唤醒在等待的消费者
    106.         pthread_rwlock_unlock(&rwlock);
    107.         sem_post(&csem); //已经生产了一个节点消费者亮灯数+1 并唤醒消费者线程中在wait的线程去消费
    108.         usleep(1000);    //每个线程在产生一个新节点后至少1ms不生产新节点
    109.     }
    110.     return NULL;
    111. }
    112. void *custom(void *arg)
    113. {
    114.     while (1)
    115.     {
    116.         //消费者不再需要单独地判断容器是否为空
    117.         //当容器为空 则消费者亮灯数为0 自然阻塞 等待生产者生产将消费者灯点亮
    118.         sem_wait(&csem);
    119.         pthread_rwlock_wrlock(&rwlock);
    120.         Node *temp = head;
    121.         head = head->next;
    122.         printf("%ld custom node %d\n", pthread_self(), temp->data);
    123.         free(temp); //释放头节点
    124.         pthread_rwlock_unlock(&rwlock);
    125.         sem_post(&psem); //已经消费了一个节点灭了一个消费者灯 需要量一个生产者灯 并唤醒生产者线程中在wait的线程去生产
    126.         usleep(1000);    //每个线程在消费一个节点后至少1ms不消费新节点
    127.     }
    128.     return NULL;
    129. }