c++11 完全公平队列实现

163 阅读 0 评论 108 点赞

我是靠谱客的博主紧张啤酒，这篇文章主要介绍c++11 完全公平队列实现，现在分享给大家，希望可以做个参考。

有时我们的服务会从多个其他子服务接收请求，不同类型的请求频率相差比较大，如果采用一个集中的队列，对到来的各个请求排队分发处理，则会出现繁忙的请求占据了队列的大部分，比较闲的请求到达后排在队尾，很长时间才会得到处理机会，出现不必要的延迟加大。这种情况下如果要加强各类型请求的响应及时性，可以采用优先队列，按照请求类型的频率加预计排队时间指定优先级，也可以使用公平队列，公平地处理各个类型的消息。

完全公平队列的原理很简单：为各个类型请求维护一个单独的子队列，维护一个链表包含所有子队列的指针；接收到请求后，根据类型找到其子队列并插入子队列末尾；分发线程分发请求时，依次遍历链表中的子队列并将此子队列移动到链表尾，如果子队列有请求，则将请求投递到工作线程执行，否则一直遍历，直到链表中的所有子队列都查询过，如果还是没有请求待处理，则进入等待。

按照上面描述，可以如下实现：unordered_map 用来保存类型和对应的子队列，list 维护着子队列的指针用来遍历。

复制代码

#ifndef __COMPLETE_FAIR_QUEUE_H_
#define __COMPLETE_FAIR_QUEUE_H_

#include <list>
#include <unordered_map>

#include <mutex>
#include <condition_variable>

template <
    class _Key,
    class _Val,
    class _Hash = std::hash<_Key>,
    class _Pred = std::equal_to<_Key> >
class cfq
{

using sub_queue = std::list<_Val>;
    using sub_queues = std::unordered_map<_Key, sub_queue, _Hash, _Pred>;
    using trival_queues = std::list<sub_queue*>;

public:

cfq() = default;
    ~cfq() = default;

void erase(const _Key &key)
    {
        std::unique_lock<std::mutex> ulk(mut_);
        auto _it = sub_queues_.find(key);
        if (_it == sub_queues_.end())
            return;
        erase_from_trival_queues(&(_it->second));
        sub_queues_.erase(_it);
    }

void push(const _Key &key, const _Val &val)
    {
        std::unique_lock<std::mutex> ulk(mut_);
        auto _it = sub_queues_.find(key);
        if (_it == sub_queues_.end())
        {
            sub_queue _sub_queue{};
            _sub_queue.push_back(val);
            auto _rit = sub_queues_.insert(std::make_pair(key, std::move(_sub_queue)));
            trival_queues_.push_back(&(_rit.first->second));
        }
        else
            _it->second.push_back(val);
        cond_.notify_one();
    }

_Val wait_and_pop()
    {
        std::unique_lock<std::mutex> ulk(mut_);
        do
        {
            auto sub_queue_size = trival_queues_.size();
            for (decltype(sub_queue_size) idx = 0; idx < sub_queue_size; ++idx)
            {
                auto _sub_queue = trival_queues_.front();
                trival_queues_.pop_front();
                trival_queues_.push_back(_sub_queue);
                if (!_sub_queue->empty())
                {
                    auto _val = _sub_queue->front();
                    _sub_queue->pop_front();
                    return _val;
                }
            }
            cond_.wait(ulk);
        } while (1);
    }

bool try_pop(_Val &val)
    {
        std::unique_lock<std::mutex> ulk(mut_);
        auto sub_queue_size = trival_queues_.size();
        for (decltype(sub_queue_size) idx = 0; idx < sub_queue_size; ++idx)
        {
            auto _sub_queue = trival_queues_.front();
            trival_queues_.pop_front();
            trival_queues_.push_back(_sub_queue);
            if (!_sub_queue->empty())
            {
                val = _sub_queue->front();
                _sub_queue->pop_front();
                return true;
            }
        }
        return false;
    }

private:

void erase_from_trival_queues(sub_queue *_sub_queue)
    {
        for (auto _it = trival_queues_.begin(); _it != trival_queues_.end(); ++_it)
        {
            if (*_it == _sub_queue)
            {
                trival_queues_.erase(_it);
                break;
            }
        }
    }

private:

std::mutex mut_;
    std::condition_variable cond_;
    sub_queues sub_queues_;
    trival_queues trival_queues_;

};

#endif //__COMPLETE_FAIR_QUEUE_H_

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
#ifndef __COMPLETE_FAIR_QUEUE_H_
#define __COMPLETE_FAIR_QUEUE_H_

#include <list>
#include <unordered_map>

#include <mutex>
#include <condition_variable>

template <
    class _Key,
    class _Val,
    class _Hash = std::hash<_Key>,
    class _Pred = std::equal_to<_Key> >
class cfq
{

    using sub_queue = std::list<_Val>;
    using sub_queues = std::unordered_map<_Key, sub_queue, _Hash, _Pred>;
    using trival_queues = std::list<sub_queue*>;

public:

    cfq() = default;
    ~cfq() = default;

    void erase(const _Key &key)
    {
        std::unique_lock<std::mutex> ulk(mut_);
        auto _it = sub_queues_.find(key);
        if (_it == sub_queues_.end())
            return;
        erase_from_trival_queues(&(_it->second));
        sub_queues_.erase(_it);
    }

    void push(const _Key &key, const _Val &val)
    {
        std::unique_lock<std::mutex> ulk(mut_);
        auto _it = sub_queues_.find(key);
        if (_it == sub_queues_.end())
        {
            sub_queue _sub_queue{};
            _sub_queue.push_back(val);
            auto _rit = sub_queues_.insert(std::make_pair(key, std::move(_sub_queue)));
            trival_queues_.push_back(&(_rit.first->second));
        }
        else
            _it->second.push_back(val);
        cond_.notify_one();
    }

    _Val wait_and_pop()
    {
        std::unique_lock<std::mutex> ulk(mut_);
        do
        {
            auto sub_queue_size = trival_queues_.size();
            for (decltype(sub_queue_size) idx = 0; idx < sub_queue_size; ++idx)
            {
                auto _sub_queue = trival_queues_.front();
                trival_queues_.pop_front();
                trival_queues_.push_back(_sub_queue);
                if (!_sub_queue->empty())
                {
                    auto _val = _sub_queue->front();
                    _sub_queue->pop_front();
                    return _val;
                }
            }
            cond_.wait(ulk);
        } while (1);
    }

    bool try_pop(_Val &val)
    {
        std::unique_lock<std::mutex> ulk(mut_);
        auto sub_queue_size = trival_queues_.size();
        for (decltype(sub_queue_size) idx = 0; idx < sub_queue_size; ++idx)
        {
            auto _sub_queue = trival_queues_.front();
            trival_queues_.pop_front();
            trival_queues_.push_back(_sub_queue);
            if (!_sub_queue->empty())
            {
                val = _sub_queue->front();
                _sub_queue->pop_front();
                return true;
            }
        }
        return false;
    }

private:

    void erase_from_trival_queues(sub_queue *_sub_queue)
    {
        for (auto _it = trival_queues_.begin(); _it != trival_queues_.end(); ++_it)
        {
            if (*_it == _sub_queue)
            {
                trival_queues_.erase(_it);
                break;
            }
        }
    }

private:

    std::mutex mut_;
    std::condition_variable cond_;
    sub_queues sub_queues_;
    trival_queues trival_queues_;

};

#endif //__COMPLETE_FAIR_QUEUE_H_