【Linux】线程安全-生产者消费者模型

生产者消费者模型

123规则

1个线程安全的队列：只要保证先进先出特性的数据结构都可以称为队列

这个队列要保证互斥（就是保证当前只有一个线程对队列进行操作，其他线程不可以同时来操作），还要保证同步，当生产者将队列中填充满了之后要通知消费者来进行消费，消费者消费之后通知生产者来进行生产。
队列起到了生产者和消费者的缓冲作用，生产者不用因为没有人消费发愁，只需要将生产的数据放到队列中即可；消费者不用因为生产者生产了大量数据而发愁，只需要正常关注正在处理的数据即可

2个角色的线程：生产者和消费者

3个规则：生产者和生产者互斥、消费者和消费者互斥、生产者和消费者互斥+同步

应用场景

比如说微信的后台程序：在不同的场景下一个进程可以是消费者也可以是生产者

优点

忙闲不均

在同一时刻可能接收消息的线程不忙而处理消息的线程一直处于工作状态

生产者和消费者解耦

生产者只关心生产，关心队列是否有空闲空间；

消费者只关心消费，关心队列中是否有数据可用。

生产者和消费者不是串行的执行（串行的处理就是当一个线程接收到消息后才可以处理消息，并且只有处理完了之后才可以发送消息，是一个串行的过程），而生产者消费者模型将生产者和消费者解耦，接收消息的一辈子就接收消息，处理消息的一辈子就处理消息，发送消息一辈子就只发送消息，不受其他线程的影响

支持高并发

同一时刻多个人发送消息这种情况是支持的，因为接收消息的线程只需要接收消息，不用干其他事情，所以接收线程接收消息的速度很快

代码模拟

采用互斥和同步实现：

#include
#include
#include
#include
#include
using namespace std;
#define THREAD_COUNT 1//生产者和消费者数量
//创建线程安全队列
class RingQueue{
public:
  RingQueue(){
    capacity = 1;
    pthread_mutex_init(&que_lock, NULL);
    pthread_cond_init(&consum_cond, NULL);
    pthread_cond_init(&product_cond, NULL);
  }
  ~RingQueue(){
    pthread_mutex_destroy(&que_lock);
    pthread_cond_destroy(&consum_cond);
    pthread_cond_destroy(&product_cond);
  }
  //往队列中放数据，生产
  void Push(int data){
    pthread_mutex_lock(&que_lock);
    while(que.size()>=capacity){
      pthread_cond_wait(&product_cond, &que_lock);
      //为什么要用while循环呢？
      //因为当生产者被唤醒后，需要再次判断队列是否可以满足生产的条件
      //生产者或者消费者都是需要在等待结束后再次判断的
    }
    que.push(data);//生产，往队列中放入数据
    cout<<"I am product: " << pthread_self() << "I product number is " << data << endl;
    pthread_mutex_unlock(&que_lock);
    pthread_cond_signal(&consum_cond);
    //生产者完成生产后唤醒消费者线程让消费者进行消费
  }
  //从队列中取数据，消费
  int Pop(){
    pthread_mutex_lock(&que_lock);
    while(que.size() <= 0){
      pthread_cond_wait(&consum_cond, &que_lock);
    }
    int data = que.front();
    que.pop();
    cout<<"I am consume: " << pthread_self() << "I consume number is " << data << endl;
    pthread_mutex_unlock(&que_lock);
    pthread_cond_signal(&product_cond);//消费者线程消费之后通知生产者来生产
    return data;
  }
  
private:
  queue<int> que;//线程安全的队列
  //给队列一把锁，保证互斥，保证同一时刻只有一个线程对队列进行操作
  pthread_mutex_t que_lock;
  //同步的条件变量，队列有元素，消息，没有元素等待，唤醒生产者
  //保证生产者在队列中没有元素的时候进行生产（插入元素）
  pthread_cond_t consum_cond;
  pthread_cond_t product_cond;
  int capacity;//队列容量，队列元素大于容量表示队满，不再往里插入元素
};
int g_val = 0;
pthread_mutex_t g_val_lock = PTHREAD_MUTEX_INITIALIZER;//静态初始化保护g_val的互斥锁
void* product_thread_start(void* arg){
  RingQueue *q = (RingQueue*)arg;
  while(1){
    pthread_mutex_lock(&g_val_lock);//获取g_val的互斥锁
    q->Push(g_val);
    g_val++;
    sleep(1);
    pthread_mutex_unlock(&g_val_lock);
  }
}
void* consum_thread_start(void* arg){
  RingQueue *q = (RingQueue*)arg;
  while(1){
    q->Pop();
  }
}
int main(){
  pthread_t consum_tid[THREAD_COUNT];
  pthread_t product_tid[THREAD_COUNT];
  RingQueue* q = new RingQueue();
  for(int i=0; i<THREAD_COUNT; ++i){
    int ret = pthread_create(&consum_tid[i], NULL, consum_thread_start, (void*)q);
    if(ret < 0){
      perror("pthread_create");
      return 0;
    }
    ret = pthread_create(&product_tid[i], NULL, product_thread_start, (void*)q);
    if(ret < 0){
      perror("pthread_create");
      return 0;
    }
  }
  for(int i=0; i<THREAD_COUNT; ++i){
    pthread_join(consum_tid[i], NULL);
    pthread_join(product_tid[i], NULL);
  }
  delete q;
  return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

执行结果：

可以看到有效的控制了生产者和消费者的消费顺序，当生产者生产一个消费者就消费一个，消费者消费后生产者接着生产