C++双队列实现生产者消费者模型demo_双队列 消费模型

传统队列是生产者线程和消费者线程从同一个队列中存取数据，必然需要互斥访问，在互相同步等待中浪费了宝贵的时间，使队列吞吐量受影响。

双缓冲队使用两个队列，将读写分离，一个队列专门用来读，另一个专门用来写，当读队列空或写队列满时将两个队列互换。这里为了保证队列的读写顺序，当读队列为空且写队列不为空时候才允许两个队列互换。

话不多说，上代码

#pragma once

#include <deque>

#include "mutex.h"//对mutex简单的区域锁封装

namespace util
{
    /*
    * 多生产者-单消费者队列
    */
    template<typename T, unsigned Size = 0>
    class FastQueue
    {
    public:

        FastQueue() : _flag(0)
        {
        }
        ~FastQueue()
        {}

        /*
        * push
        */
        inline void push(const T& elem)
        {
            _mutex.lock();
            if (_flag == 0)
                _containerB.push_back(elem);
            else
                _containerA.push_back(elem);
            _mutex.unlock();
        }

        /*
        * pop
        */
        bool pop(T& elem)
        {
            if (_flag == 0)
            {
                if (!_containerA.empty())
                {
                    elem = _containerA.front();
                    _containerA.pop_front();
                    return true;
                }
                else
                {
                    _mutex.lock();
                    if (!_containerB.empty())
                    {
                        _flag = 1;
                        _mutex.unlock();

                        elem = _containerB.front();
                        _containerB.pop_front();
                        return true;
                    }
                    _mutex.unlock();
                }
            }
            else
            {
                if (!_containerB.empty())
                {
                    elem = _containerB.front();
                    _containerB.pop_front();
                    return true;
                }
                else
                {
                    _mutex.lock();
                    if (!_containerA.empty())
                    {
                        _flag = 0;
                        _mutex.unlock();

                        elem = _containerA.front();
                        _containerA.pop_front();
                        return true;
                    }
                    _mutex.unlock();
                }
            }

            return false;
        }

        /*
        * 移除未使用的空间
        */
        inline void reserved()
        {
            _mutex.lock();
            _containerA.shrink_to_fit();
            _containerB.shrink_to_fit();
            _mutex.unlock();
        }

        inline unsigned getSize()
        {
            _mutex.lock();
            unsigned int size = 0;
            size = _containerA.size() + _containerB.size();
            _mutex.unlock();
            return size;
        }

        inline unsigned getSizeR()
        {
            if (_flag == 0)
                return (unsigned)_containerA.size();
            else
                return (unsigned)_containerB.size();
        }

        inline unsigned getSizeW()
        {
            _mutex.lock();
            unsigned int size = 0;
            if (_flag == 0)
                size = _containerB.size();
            else
                size = _containerA.size();
            _mutex.unlock();
            return size;
        }

        inline int getFlag(){ return _flag; }

    private:
        std::deque<T> _containerA;
        std::deque<T> _containerB;
        int _flag;         //读线程切换队列状态
        Mutex _mutex;
    };

}
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140