linux：生产者消耗者模子

   个人主页 ： 个人主页
个人专栏 ： 《数据结构》 《C语言》《C++》《Linux》
  
  

---

前言

本文是对于生产者消耗者模子的知识总结

---

一、生产者消耗者模子

生产者消耗者模子就是通过一个容器来解决生产者消耗者的强耦合问题。生产者和消耗者相互之间不直接通讯，而是通过之间的容器来进行通讯，所以生产者生产完数据之后不消等待消耗者处理，直接交给容器，消耗者不找生产者要数据，而是直接从容器中取数据，壅闭队列就相称于一个缓冲区，均衡了生产者和消耗者的处理能力，其中这个容器就是用于生产者和消耗者之间解耦的
(强耦合是指两个或多个系统，组件或模块之间存在紧密依赖关系。)

---

特点：

• 三种关系：生产者与生产者之间(互斥)，消耗者与消耗者之间(互斥)，生产者与消耗者之间(互斥 && 同步)
  
• 两种脚色：生产者和消耗者
  
• 一个交易(通讯)场所：一个容器(一段内存空间)
  

由于我们是多个线程访问同一个容器，那必然会导致数据不一致的问题，所以我们必要对该临界资源加锁，所以生产者与生产者之间，消耗者与消耗者之间，生产者与消耗者之间都是互斥的。
又由于容器大概为空(满)，此时消耗者(生产者)还不停在临界区申请锁，又因没有数据(空间)而释放锁，从而不断申请锁释放锁，导致生产者(消耗者)的饥饿问题。此时我们就必要生产者与消耗者之间的同步。
对于2,3两点，这很好明白不解释。
我们编写生产者消耗者模子的本质就是对以上三点的维护。(互斥包管数据安全，同步包管服从)

---

长处：

• 解耦
  
• 支持并发
  
• 支持忙闲不均
  

对于第一点，不就是生产者与消耗者通过容器来解耦提拔服从(如果没有这个容器，则生产者生产完数据，就必须等待消耗者来接受处理数据，不能立即继续生产数据)。
对于第二点，当生产者在生产数据时，消耗者也同时在处理数据
对于第三点，当生产者生产数据的速度凌驾消耗者的处理能力时，容器可以起到缓存的作用，将多余的数据暂时存储，等待消耗者有空闲时再进行处理。如果消耗者处理数据的能力凌驾生产者时，同理。
二、基于壅闭队列的生产者消耗者模子

在多线程编程中壅闭队列(Blocking Queue)是一种常用于实现生产者和消耗者模子的数据结构。

• 队列为空时，从队列中取数据将被壅闭，直到队列中有数据时被唤醒。
  
• 队列为满时，向队列中放入数据将被壅闭，直到队列中有数据取出被唤醒。
  

代码实现

下面是一个单生成单消耗模子

LockGuard.hpp 文件 将加锁释放锁，交给一个对象处理，当对象创建加锁，对象烧毁释放锁

1. #pragma once
2. #include <pthread.h>
4. class Mutex
5. {
6. public:
7.     Mutex(pthread_mutex_t *mutex):_mutex(mutex)
8.     {}
10.     void Lock()
11.     {
12.         pthread_mutex_lock(_mutex);
13.     }
15.     void UnLock()
16.     {
17.         pthread_mutex_unlock(_mutex);
18.     }
20.     ~Mutex()
21.     {}
22. private:
23.     pthread_mutex_t *_mutex;
24. };
26. class LockGuard
27. {
28. public:
29.     LockGuard(pthread_mutex_t *mutex): _lock(mutex)
30.     {
31.         _lock.Lock();
32.     }
34.     ~LockGuard()
35.     {
36.         _lock.UnLock();
37.     }
38. private:
39.     Mutex _lock;
40. };

复制代码

Blockqueue.hpp 文件

1. #pragma once
2. #include <iostream>
3. #include <queue>
4. #include <pthread.h>
5. #include "LockGuard.hpp"
7. using namespace std;
8. const int CAPACITY = 5;
10. template<class T>
11. class BlockQueue
12. {
13. public:
14.     BlockQueue(int cap = CAPACITY):_capacity(cap)
15.     {
16.         pthread_mutex_init(&_mutex, nullptr);
17.         pthread_cond_init(&_p, nullptr);
18.         pthread_cond_init(&_c, nullptr);
19.     }
21.     bool isFull()
22.     {
23.         return _bq.size() == _capacity;
24.     }
26.     void Push(const T &in)
27.     {
28.         LockGuard mutex(&_mutex);
29.         //pthread_mutex_lock(&_mutex);
30.         while(isFull())
31.         {
32.             pthread_cond_wait(&_p, &_mutex);
33.         }
35.         _bq.push(in);
36.         // 唤醒策略为 生产一个，消费一个
37.         pthread_cond_signal(&_c);
39.         //pthread_mutex_unlock(&_mutex);
40.     }
42.     bool isEmpty()
43.     {
44.         return _bq.size() == 0;
45.     }
47.     void Pop(T *out)
48.     {
49.         LockGuard mutex(&_mutex);
50.         //pthread_mutex_lock(&_mutex);
51.         while(isEmpty())
52.         {
53.             pthread_cond_wait(&_c, &_mutex);
54.         }
56.         *out = _bq.front();
57.         _bq.pop();
58.         // 唤醒策略为 消费一个，生产一个
59.         pthread_cond_signal(&_p);
61.         //pthread_mutex_unlock(&_mutex);
62.     }
64.     ~BlockQueue()
65.     {
66.         pthread_mutex_destroy(&_mutex);
67.         pthread_cond_destroy(&_p);
68.         pthread_cond_destroy(&_c);
69.     }
70. private:
71.     queue<T> _bq;
72.     int _capacity;
74.     pthread_mutex_t _mutex;
75.     pthread_cond_t _p;
76.     pthread_cond_t _c;
77. };

复制代码

Task.hpp 文件

1. #pragma once
2. #include <string>
4. const char *opers = "+-*/%";
6. enum
7. {
8.     ok = 0,
9.     div_zero,
10.     mod_zero
11. };
13. class Task
14. {
15. public:
16.     Task()
17.     {}
19.     Task(int x, int y, char op) : _data_x(x), _data_y(y), _oper(op)
20.     {
21.         _code = ok;
22.     }
24.     void Run()
25.     {
26.         switch (_oper)
27.         {
28.         case '+':
29.             _result = _data_x + _data_y;
30.             break;
31.         case '-':
32.             _result = _data_x - _data_y;
33.             break;
34.         case '*':
35.             _result = _data_x * _data_y;
36.             break;
37.         case '/':
38.             {
39.                 if(_data_y == 0)
40.                 {
41.                     _code = div_zero;
42.                 }
43.                 else
44.                 {
45.                     _result = _data_x / _data_y;
46.                 }
47.             }
48.             break;
49.         case '%':
50.             {
51.                 if(_data_y == 0)
52.                 {
53.                     _code = mod_zero;
54.                 }
55.                 else
56.                 {
57.                     _result = _data_x % _data_y;
58.                 }
59.             }
60.             break;
61.         default:
62.             break;
63.         }
64.     }
66.     void operator()()
67.     {
68.         Run();
69.     }
71.     std::string PrintTask()
72.     {
73.         std::string ret = std::to_string(_data_x);
74.         ret += _oper;
75.         ret += std::to_string(_data_y);
77.         ret += "=?";
78.         return ret;
79.     }
81.     std::string PrintResult()
82.     {
83.         std::string ret = std::to_string(_data_x);
84.         ret += _oper;
85.         ret += std::to_string(_data_y);
87.         ret += "=";
88.         if(_code == ok)
89.         {
90.             ret += std::to_string(_result);
91.         }
92.         else
93.         {
94.             ret += "?";
95.         }
97.         ret += "[";
98.         ret += std::to_string(_code);
99.         ret += "]";
101.         return ret;
102.     }
104.     ~Task()
105.     {}
107. private:
108.     int _data_x;
109.     int _data_y;
110.     char _oper;
112.     int _result;
113.     int _code; // 错误码
114. };

复制代码

Main.cc 文件

1. #include <iostream>
2. #include <pthread.h>
3. #include <unistd.h>
4. #include <ctime>
5. #include <string.h>
6. #include "BlockQueue.hpp"
7. #include "Task.hpp"
8. using namespace std;
11. void *producer(void *args)
12. {
13.     BlockQueue<Task> *bq = static_cast<BlockQueue<Task> *>(args);
14.     // 产生任务
15.     while (true)
16.     {
17.         int x = rand() % 10 + 1;
18.         int y = rand() % 10 + 1;
19.         char oper = opers[rand() % strlen(opers)];
21.         Task task(x, y, oper);
22.         cout << "producer: " << task.PrintTask() << endl;
23.         bq->Push(task);
24.         sleep(1);
25.     }
27.     return nullptr;
28. }
30. void *consumer(void *args)
31. {
32.     // usleep(1000);
33.     BlockQueue<Task> *bq = static_cast<BlockQueue<Task> *>(args);
34.     // 获取任务，处理任务
35.     while (true)
36.     {
37.         if (bq->isFull())
38.         {
39.             Task task;
40.             bq->Pop(&task);
42.             task();
43.             cout << "consumer: " << task.PrintResult() << endl;
44.             //sleep(1);
45.         }
46.     }
47. }
49. int main()
50. {
51.     srand(time(nullptr) ^ getpid());
52.     BlockQueue<Task> bq;
54.     pthread_t p;
55.     pthread_create(&p, nullptr, producer, (void *)&bq);
57.     pthread_t c;
58.     pthread_create(&c, nullptr, consumer, (void *)&bq);
60.     pthread_join(p, nullptr);
61.     pthread_join(c, nullptr);
62.     return 0;
63. }

复制代码

那怎样将这个单生产单消耗该为多生产多消耗呢？由于多生产多消耗本质也是多个线程访问临界资源，那我们单生产和单消耗不也是多个线程访问临界资源吗，所以我们不必要对BlockQueue.hpp文件进行修改，只必要在main函数中，创建多个生产者和消耗者即可。

1. #include <iostream>
2. #include <pthread.h>
3. #include <unistd.h>
4. #include <ctime>
5. #include <string.h>
6. #include "BlockQueue.hpp"
7. #include "Task.hpp"
8. using namespace std;
10. template <class T>
11. class ThreadData
12. {
13. public:
14.     ThreadData(pthread_t tid, const string threadname, BlockQueue<T> *bq)
15.         : _tid(tid), _threadname(threadname), _bq(bq)
16.     {}
18. public:
19.     pthread_t _tid;
20.     string _threadname;
21.     BlockQueue<T>* _bq;
22. };
24. void *producer(void *args)
25. {
26.     ThreadData<Task> *data = static_cast<ThreadData<Task> *>(args);
27.     // 产生任务
28.     while (true)
29.     {
30.         int x = rand() % 10 + 1;
31.         int y = rand() % 10 + 1;
32.         char oper = opers[rand() % strlen(opers)];
34.         Task task(x, y, oper);
35.         cout << data->_tid << ", " << data->_threadname <<": " << task.PrintTask() << endl;
36.         data->_bq->Push(task);
37.         sleep(1);
38.     }
40.     return nullptr;
41. }
43. void *consumer(void *args)
44. {
45.     // usleep(1000);
46.     ThreadData<Task> *data = static_cast<ThreadData<Task> *>(args);
47.     // 获取任务，处理任务
48.     while (true)
49.     {
50.         if (data->_bq->isFull())
51.         {
52.             Task task;
53.             data->_bq->Pop(&task);
55.             task();
56.             cout << data->_tid << ", " << data->_threadname << ": " << task.PrintResult() << endl;
57.             // sleep(1);
58.         }
59.     }
60. }
64. int main()
65. {
66.     srand(time(nullptr) ^ getpid());
67.     BlockQueue<Task> bq;
69.     pthread_t p1;
70.     ThreadData<Task> data1(p1, "product-1", &bq);
71.     pthread_create(&p1, nullptr, producer, (void *)&data1);
73.     pthread_t p2;
74.     ThreadData<Task> data2(p2, "product-2", &bq);
75.     pthread_create(&p2, nullptr, producer, (void *)&data2);
77.     pthread_t c1;
78.     ThreadData<Task> data3(c1, "consumer-1", &bq);
79.     pthread_create(&c1, nullptr, consumer, (void *)&data3);
80.     pthread_t c2;
81.     ThreadData<Task> data4(c2, "consumer-2", &bq);
82.     pthread_create(&c2, nullptr, consumer, (void *)&data4);
84.     pthread_join(p1, nullptr);
85.     pthread_join(p2, nullptr);
87.     pthread_join(c1, nullptr);
88.     pthread_join(c2, nullptr);
90.     return 0;
91. }

复制代码

---

总结

以上就是我对于线程同步的总结。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。