linux多线程3-生产者消费者模型

1、生产者消费者模型：假设有一筐饼，这时候生产者源源不断生产饼放进饼框中，此时有三个消费者（用三个线程代替），这时，三个人都想抢饼吃，这时候，当没有饼的时候他们就应该陷入阻塞等待状态，当有饼的时候，他们就会被告知饼来了，然后竞争获取新的饼，这时候，除了加锁之外，还需要一个重要的事情，就是需要被通知到饼来了这个情况，从而解除阻塞等待的状态，这个东西就叫做条件变量；

2、条件变量：可以实现阻塞，但并非锁，要和互斥量组合使用；

　　初始化一个条件变量：int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr);

　　　　　　　　　　　　pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

　　　　　　　　　　　　可以定义一个变量通过函数赋值或者直接给一个条件变量进程初始化赋值；

　　销毁一个条件变量：int pthread_cond_destroy(pthread_cond_t *cond);

　　条件变量阻塞等待：1）int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);

　　　　　　　　　　　2）int pthread_cond_timedwait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex, const struct timespec *restrict abstime);

　　　　　　　　　　　方法一是一直阻塞等待，方法二是超时等待，即超过给定的时间就不会进行等待了；

　　　　　　　　　　　阻塞等待的方式是先将锁释放，这样所有线程就都会阻塞到条件变量上。　

　　　　　　　　　　　方法二时间的结构体：struct timespec{

　　　　　　　　　　　　　　　　　　　　　　　　　　　　time_t tv_sec; /*seconds 秒*/

　　　　　　　　　　　　　　　　　　　　　　　　　　　　long tv_nsec;

　　　　　　　　　　　　　　　　　　　　　　　　　　　　} /*纳秒*/ 　tv_sec是绝对时间，设置时需要time（NULL）+ time；

　　唤醒阻塞在条件变量cond上的线程：1）int pthread_cond_broadcast(pthread_cond_t *cond);

　　　　　　　　　　　　　　　　　　2）int pthread_cond_signal(pthread_cond_t *cond);

　　　　　　　　　　　　　　　　　　方法一是唤醒阻塞在条件变量cond上的全部线程，方法二是唤醒至少一个阻塞在条件变量cond上的线程；

3、使用条件变量实现生产者消费者模型：

 1 #include <iostream>
 2 #include <unistd.h>
 3 #include <pthread.h>
 4 #include <ctime>
 5 #include <cstdlib>
 6 using namespace std;
 7 int beginnum = 1000;
 8 
 9 pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;  //初始化互斥量锁
10 pthread_cond_t cond = PTHREAD_COND_INITIALIZER;　　//初始化一个条件变量
11 typedef struct _ProdInfo{
12     int num;
13     struct _ProdInfo *next;
14 } ProdInfo;    //定义一个链表结构体，用于生产者和消费者使用
15 
16 ProdInfo *head = NULL;
17 void *thr_producter(void *arg)  //定义生产者
18 {
19     srand(time(NULL));
20     while(1)
21     {
22         pthread_mutex_lock(&mutex);    //加锁
23         ProdInfo *info = new ProdInfo;
24         info->num = beginnum++;
25         info->next = head;
26         head = info;
27         pthread_mutex_unlock(&mutex);   //释放锁
28         pthread_cond_broadcast(&cond);  //生产完成，通知所有线程生产OK
29         sleep(rand() % 3);
30     }
31     return NULL;
32 }
33 
34 
35 void *thr_consumer(void *arg)  //消费者
36 {
37     ProdInfo *mytype = NULL;
38     while(1)
39     {
40         pthread_mutex_lock(&mutex);  /
41         while(head == NULL)
42         {
43             pthread_cond_wait(&cond, &mutex);  // 此处为消费者阻塞在条件变量，当head为空时，退出阻塞状态，同时加上锁。这里需要注意用while而不是if，因为可能其他线程同时解除阻塞，这样就会导致异常。
44         }
45         mytype = head;
46         head = head->next;
47         pthread_mutex_unlock(&mutex);
48         cout << __FUNCTION__ << "thread--tid = " << pthread_self() << " typeNum = " << mytype->num << endl;
49         delete mytype;
50         sleep(rand() % 2);
51     }
52     return NULL;
53 }
54 
55 
56 int main(int argc, char *argv[])
57 {
58     pthread_t tid[3];
59     pthread_create(&tid[0], NULL, thr_producter, NULL);
60     pthread_create(&tid[1], NULL, thr_consumer, NULL);
61     pthread_create(&tid[2], NULL, thr_consumer, NULL);
62     pthread_join(tid[0], NULL);
63     pthread_join(tid[1], NULL);
64     pthread_join(tid[2], NULL);
65     pthread_mutex_destroy(&mutex);
66     pthread_cond_destroy(&cond);
67     return 0;
68 }

4、信号量：之前说使用互斥量和条件变量的组合来实现的生产者消费者模型是有局限性的，每次只能生产一个；假设有三个厕所，如果使用互斥量，那么每次进入一个厕所，其他两个厕所也没办法进去，这时候就大大降低了效率，所以，我们通过信号量的方式来通知人，还有可用的厕所；由于互斥锁的粒度比较大，如果我们希望在多个线程间对某一对象的部分数据进行共享，使用互斥锁是没办法实现的，只能讲整个数据锁住，这样虽然达到了多线程操作共享数据时保证数据正确性的目的，但是却导致线程并发性下降，使其与单进程没有差异。信号量是一种折中的方式，既保证同步，也一定程度上提高了线程的并发。

　　#include <semaphore.h>

　　初始化信号量：int sem_init(sem_t *sem, int pshared, unsigned int value);

　　　　　　　　　sem：传入的信号量参数，对其进行初始化；

　　　　　　　　　pshared：为0时表示多线程，非0表示多进程；

　　　　　　　　　value：代表并发的个数，可以理解为锁的个数；

　　销毁信号量：int sem_destroy(sem_t *sem);

　　信号量申请：int sem_wait(sem_t *sem);

　　　　　　　　int sem_trywait(sem_t *sem);

　　　　　　　　int sem_timedwait(sem_t *sem, const struct timespec *abs_timeout);

　　　　　　　　struct timespec {

　　　　　　　　　　time_t tv_sec; /* Seconds */
　　　　　　　　　　long tv_nsec; /* Nanoseconds [0 .. 999999999] */
　　　　　　　　};

　　　　　　　　申请信号量，申请成功，value--；当信号量为0时，会阻塞；

　　信号量释放：int sem_post(sem_t *sem);

　　　　　　　　释放信号量，value++;

5、使用信号量实现生产者消费者模型：

 1 #include <iostream>
 2 #include <unistd.h>
 3 #include <pthread.h>
 4 #include <semaphore.h>
 5 #include <cstdlib>
 6 using namespace std;
 7 #define MAX_SEM_NUM 5
 8 sem_t pro_value;
 9 sem_t cons_value;
10 int startNum = 100;
11 int start_index = 0;
12 int goods[MAX_SEM_NUM];
13 void *thr_producter(void *arg)
14 {
15     int i = 0;
16     while(1)
17     {
18         sem_wait(&pro_value);  // 相当于生产者value自减操作
19         cout << __FUNCTION__ << ": pruduce_num = " << startNum << " pid = " << pthread_self() << endl;
20         goods[i++ % MAX_SEM_NUM] = startNum++;
21         sem_post(&cons_value); // 消费者value+1
22         sleep(rand() % 3);
23     }
24     return NULL;
25 }
26 
27 void *thr_consumer(void *arg)
28 {
29     int i = 0;
30     int res;
31     while(1)
32     {
33         sem_wait(&cons_value);
34         res = goods[(start_index++) % MAX_SEM_NUM];
35         cout << __FUNCTION__ << ": consume_num = " << res << " pid = " << pthread_self() << endl;
36         sem_post(&pro_value);
37         sleep(rand() % 3);
38     }
39     return NULL;
40 }
41 
42 
43 int main(int argc, char *argv[])
44 {
45     pthread_t tid[3];
46     sem_init(&pro_value, 0, MAX_SEM_NUM);
47     sem_init(&cons_value, 0, 0);
48     pthread_create(&tid[0], NULL, thr_producter, NULL);
49     pthread_create(&tid[1], NULL, thr_consumer, NULL);
50     pthread_create(&tid[2], NULL, thr_consumer, NULL);
51     pthread_join(tid[0], NULL);
52     pthread_join(tid[1], NULL);
53     pthread_join(tid[2], NULL);
54     sem_destroy(&pro_value);
55     sem_destroy(&cons_value);
56     return 0;
57 }

https://github.com/LeechanX/Ring-Log/blob/master/example/multi-thread-example.cc