C++ 多线程 线程安全队列设计

这是看《C++并发编程实战》这本书学的，这里我要为这本书辟谣一下，虽然是这本书前面翻译得很烂，但是从第6章开始，应该是换了个人翻译，虽然还是能难懂，但是难懂的是代码逻辑，而不是语言逻辑；

实现，我先说明一下我自己的一个感悟，即对大多数线程错误的感悟：

1：我设定一个“信息有效期”这个概念，这个概念是值我从一个信息源获取一个信息，这个信息相对这个信息源是正确的时间段，比如一个线程获取一个队列是否为空，如果得到的结果是true，那么从获取这个信息到队列中添加一个元素这段时间就是获取到的信息的有效期；

2：大多数线程错误都是来源于信息有效期已经过去，我们知道，算法他的每一个步骤都是依赖之前的步骤的，如果前一个步骤的结果到实现依赖这个步骤的步骤时是错误的，那么不能说绝对，只能说是很可能这个步骤执行完后的结果也是错误的，那么这个“错误链”就会传递下去，如果没有正确的异常处理，那么这个程序很可能就会崩溃；

3：怎么避免信息有效期过去？

    -：获得的信息有效期够长

    -：利用锁来将信息有效期延长；

    -：获得的信息正确与否不重要，我们只关系信息本身【这个也可能是无锁并发的实现思想之一】

4：细粒度，如果我们要一个多线程安全的数据结构效率高的话，我们一般是需要控制好锁保护的数据范围，一般来说，一个锁保护的数据越多，那么这个数据结构并发访问的效率就越低，相反则反，另外一个方面，当一个操作需要的锁越多，那么并发访问的效率就越低，相反则反；

粗粒度队列的设计

一般我们是怎么设计一个队列的呢，利用锁将单线程安全队列变为多线程安全队列？

直接用一个单链表就行了；

我们只需要用两个互斥量来保护head和tail的是串行的即可；

template<typename T>
class queue
{
private:
    struct node
    {
        T data;
        std::unique_ptr<node> next;
        node(T data_) :
            data(std::move(data_))
        {}
    };
    std::unique_ptr<node> head; // 1
    node* tail; // 2
    mutex head_mutex, tail_mutex;
public:
    queue()
    {}
    queue(const queue& other) = delete;
    queue& operator=(const queue& other) = delete;
    std::shared_ptr<T> try_pop()
    {
        lock_guard<mutex> lock_g(head_mutex);
        if (!head){
            return std::shared_ptr<T>();
        }
        std::shared_ptr<T> const res(
            std::make_shared<T>(std::move(head->data)));
        std::unique_ptr<node> const old_head = std::move(head);
        head = std::move(old_head->next); // 3
        return res;
    }
    void push(T new_value)
    {
        lock_guard<mutex> lock_g(tail_mutex);
        lock_guard<mutex> lock_g(head_mutex);
        std::unique_ptr<node> p(new node(std::move(new_value)));
        node* const new_tail = p.get();
        if (tail){
            tail->next = std::move(p); // 4
        }
        else{
            head = std::move(p); // 5
        }
        tail = new_tail; // 6
    }
};

这里我们可以看到，每一个锁在保护数据的数量上，已经是非常少了【每一个锁只保护了一个节点，这已经是非常理想的状态了】，那么我们可以看到在try_pop操作当中，我们只依赖与tail节点，而在push操作当中，我们依赖head和tail两个节点，那么我们能不能进行修改，让push只依赖一个节点呢？答案是：能；

细粒度队列的设计

我们只需要一个虚拟节点即可，这个节点我们保证一直在队列的最后，且队列稳定的时候，tail恒定指向这个虚拟节点，这样的话，当我们push的时候，我们就可以不用依赖与head节点了，因为无论是空队列还是非空队列，在执行push操作的前后，head的指向都不变，这意味着我们可以不访问head节点；

我们在构造队列的时候，我们创建一个虚拟节点，然后让head和tail都指向这个节点，要稍微注意的是，判空条件不再是head==nullptr了，而是head.get()==tail；

看具体代码：

class Queue {
public:
    class node {
    public:
        shared_ptr<int> data;
        unique_ptr<node> next;
    };
private:
    unique_ptr<node> head;
    node* tail;
    mutex tail_mtx, head_mtx;

    unique_ptr<node> pop_head() {
        std::lock_guard<mutex> lock_g(head_mtx);
        if (head.get() == get_tail()) {
            return nullptr;
        }
        unique_ptr<node> old_head = std::move(head);
        head = std::move(old_head->next);
        return old_head;
    }
    node* get_tail() {
        std::lock_guard<mutex> lock_g(tail_mtx);
        return tail;
    }

public:
    Queue()
        :head(std::make_unique<node>()), tail(head.get()){}
    void push(int new_date) {
        shared_ptr<int> date = std::make_shared<int>(new_date);
        unique_ptr<node> new_tail = std::make_unique<node>();
        node* temp = new_tail.get();
        std::lock_guard<mutex> lock_g(tail_mtx);
        tail->data = date;
        tail->next = std::move(new_tail);
        tail = temp;
    }

    std::shared_ptr<int> try_pop() {
        unique_ptr<node> ptr = pop_head();
        return ptr == nullptr ? nullptr : ptr->data;
    }
};

那么好，这个时候这个队列的并发访问性已经是非常高了【主要是队列能够操作的自由度是真不高】

那么好，我们现在再提出一个需求，有的时候，我们一个线程必须要从一个队列中获取一个元素，才能进行下面的操作【比如线程池】，而我们不希望用一个while循环来进行检查队列中是否有元素以供获取，这个时候我们可以通过环境变量来进行，但是我们希望这个wait的过程与实际的业务代码是低耦合的，所以我们要把，这个逻辑封装到队列中；

这里我们可以通过两种方式来获取队列中的值，一个是用引用hook，一个是直接通过返回值来获取；

当然，我们用引用hook时，就不免会用到客户端程序员的带代码，移动拷贝结果的时候可能会报错，这个时候需要对具体的异常进行捕获，为了避免这个异常，我们设计节点用share_ptr来存储结果，因为share_ptr内部在创建对象的时候有异常处理，但是这样做的坏处是我们分配的结果必须存储到堆中，这为push时创建结果的过程增加了不少的时间开销；

下面是具体代码：

template<typename T>
class queue_plus {
private:
    class node {
    public:
        std::shared_ptr<T> date;
        std::unique_ptr<node> next;
    };
    mutex head_mtx, tail_mtx, size_mtx;
    unique_ptr<node> head;
    node* tail;
    condition_variable cond;
    
    node* get_tail() {
        std::lock_guard<mutex> lock_g(tail_mtx);
        return tail;
    }

    shared_ptr<T> pop_head() {
        if (head->get() == get_tail()) {
            return nullptr;
        }
        shared_ptr<T> date=std::move(head->date);
        head = std::move(head->next);
        return date;
    }

    unique_lock<std::mutex> wait_for_notify() {
        std::unique_lock<std::mutex> head_lock(head_mtx);
        cond.wait(head_lock, [&] {head.get() != get_tail(); });
        return head_lock;
    }

    void wait_for_date(T& value) {
        std::unique_lock<std::mutex> head_lock(wait_for_notify());
        shared_ptr<T> date = pop_head();
        if (date != nullptr) {
            value = std::move(*date);
        }
    }
    shared_ptr<T> wait_for_date() {
        std::unique_lock<std::mutex> head_lock(wait_for_notify());
        return pop_head();
    }
public:

    void wait_and_pop(T& vlaue) {
        wait_for_date(value);
    }

    shared_ptr<T> wait_and_pop() {
        return wait_for_date();
    }

};