这篇文章主要介绍一个能够高效使用内存并且线程安全的数据结构——映射队列。这个队列是在写并发 B+ 树的时候想出来的。

    这个队列最大的特点在于它在不伤害性能的前提下大规模地减少了内存的分配以及释放，从而非常优雅的内存使用。

    这个队列的建议使用环境

    • 队列元素不适合深拷贝（deep copy），不适合深拷贝的可能原因有，元素占用内存较大，或者元素使用内存不定
    • 程序对内存的使用很敏感

    传统线程安全队列存在的问题

    不定长队列：不定长队列存在的问题是如果消费者的速度远小于生产者的速度，那么会导致大量的元素堆积，造成大量的内存被浪费。

    循环队列：循环队列存在的问题是队列中各个 slot 的元素使用时间是不均衡的，也就是说可能 slot 1 使用的时间比较长，但是 slot 2 可能使用的时间比较短，导致 slot 2 不能被后续线程提前使用。

    所以映射队列解决的问题不光是能够更少地使用内存，同时能够让队列中的资源能够更加高效地被利用。一个非常自然的想法是维护一个自己的内存池，每份内存加个标签表示是否正在被使用。也就是说我们插入索引时首先向内存池申请，寻找到一份空闲的内存，然后将要被放入到队列中的元素拷贝到这份内存，最后放入循环队列等待调用。

    这个方法我尝试过，但行不通。为什么，因为每个线程执行任务的花费的时间可能差距很大，这会造成内存池的空洞。为什么？因为没有线程能保证先开始的先结束，尤其是在遇到 B+ 树页面分裂的时候，所以我们自己的内存池在使用一段时间后会可能变成下面这个样子。

    img

    也就是说我们每次获取索引内存时还需要对内存池来一次遍历，这显然行不通。

    为了解决以上的问题，映射队列横空出世。

    • 什么是映射队列？ 映射队列是通过三个指针维护两个映射数组的循环队列，它同时起到了线程安全队列以及内存资源管理器的作用。以下是核心结构：
    1. template<typename Fuck>
    2. class BoundedMappingQueue
    3. {
    4.   private:
    5.     vector<Fuck> elements_;
    6.     vector<int>  avail_;
    7.     vector<int>  work_; 
    8.     int          front_;
    9.     int          avail_back_;
    10.     int          work_back_;
    11. }

    elements_：存放实际队列元素

    avail_：空闲元素的位置，avail_[0]表示第一个空闲元素在elements_中的位置，也就是说，elements_[avail_[0]]才表示空闲元素的具体位置

    work_：使用中元素的位置，和avail_相对应

    front_：avail_和work_的指针，avail_[front_]表示下一个空闲元素的位置，work_[front_]表示可以被使用的元素的位置

    avail_back_：avail_的指针，avail_[avail_back_]表示最后一个空闲元素的位置

    work_back_：work_的指针，work_[work_back_]同样表示可以被使用的元素的位置

    • 这个队列如何工作？

    首先，队列会这样被初始化。假设队列有N个元素，则avail_中的元素依次初始化为0，１，……N-1，而work_中的元素都初始化为-1。

    当我们需要往队列中插入某个元素时，我们首先从avail_中获取元素的实际位置，然后取得实际元素并进行我们需要的操作（初始化、赋值等等），然后我们更新avail_，将avail_[front_]赋值给work_[front_]，之后将avail_[front_]赋值为-1，并且将front_+1。

    当我们的线程池中的线程发现work_[work_back_]大于等于0时，则结束等待，获取实际元素，并且记录下此时work_[work_back_]的值，同时将work_[work_back_]赋值为-1，并且work_back_+1，然后进行具体工作的执行。

    当工作执行完毕后，利用我们之前记录的work_[work_back_]的值，将其赋值给avail_[avail_back_]，然后avail_back_+1。

    至此一次完整的放置任务、执行任务的过程就结束了。

    这个队列成功地解决了所有在上面被描述过的问题：

    1. 在初始化完毕后不需要任何new和delete
    2. 队列占用内存很少
    3. 不会出现执行任务的线程时间差异而导致的队列内存的空洞
    4. 不损失性能

    我们已经介绍完了为什么会有映射队列以及它相对于不定长队列、循环队列的优势。评论里有同学说可以用链表，我发现一个空闲链表加上一个工作链表确实可以发挥映射队列的功能，而且很好理解也很好实现，但是一次完整的操作需要加锁4次，解锁4次，而映射队列只需要3次，所以又发现了一个映射队列的优点。

    与传统的队列不同，映射队列有4个API，分成两对。分别是：

    1. template<typename Fuck>
    2. BoundedMappingQueue
    3. {
    4.   public:
    5.     Fuck* Get();
    7.     void Push();
    9.     Fuck* Pop(int *pos);
    11.     void Put(int pos);
    12. };

    其中Get和Push由用户调用，Pop和Put由线程池调用。

    1. // 用户
    2. Fuck *shit = queue.Get();
    3. // 对shit进行初始化或赋值
    4. // 初始化或赋值需要快
    5. // 因为此时队列处于加锁状态
    6. queue.Push();
    8. // 线程池
    9. int pos;
    10. Fuck *shit = queue.Pop(&pos);
    11. (*shit)();// 干活
    12. queue.Put(pos);

    下面我们具体介绍映射队列的每个API。

    1. template<typename Fuck>
    2. Fuck* Get()
    3. {
    4.     mutex_.Lock();
    5.     // 等待空闲的元素
    6.     while (avail_[front_] < 0)
    7.         empty_.Wait(mutex_);
    8.     return queue_[avail_[front_]];
    9.     // 不解锁，因为我们对参数进行初始化
    10.     // 后会立刻调用Push
    11. }
    13. template<typename Fuck>
    14. void Push()
    15. {
    16.     // 将初始化完毕的参数位置传给工作槽
    17.     work_[front_] = avail_[front_];
    18.     // 从空闲槽将元素位置删除
    19.     avail_[front_] = -1;
    20.     if (++front_ == capacity_)
    21.         front_ = 0;
    23.     mutex_.Unlock();
    24.     // 唤醒工作线程
    25.     ready_.Signal();
    26. }
    1. template<typename Fuck>
    2. Fuck* Pop(int *pos)
    3. {
    4.     mutex_.Lock();
    5.     // 等待可以工作的元素
    6.     while (work_[work_back_] < 0)
    7.         ready_.Wait(mutex_);
    9.     // 获取可以工作的元素位置并保存于pos
    10.     // 我们需要保存pos因为在完成工作后
    11.     // 需要将元素位置重新放入空闲槽
    12.     *pos = work_[work_back_];
    14.     Fuck *shit = queue_[*pos];
    16.     // 从工作槽中将元素位置删除
    17.     work_[work_back_] = -1;
    18.     if (++work_back_ == capacity_)
    19.         work_back_ = 0;
    21.     mutex_.Unlock();
    22.     return shit;
    23. }
    26. template<typename T>
    27. void Put(int pos)
    28. {
    29.     mutex_.Lock();
    30.     // 将空闲元素的位置放入空闲槽
    31.     avail_[avail_back_] = pos;
    32.     if (++avail_back_ == capacity_)
    33.         avail_back_ = 0;
    35.     mutex_.Unlock();
    36.     // 唤醒入列线程
    37.     empty_.Signal();
    38. }

    这个队列的源码 https://github.com/UncP/aili，在 blink/mapping_arry.* 中。