C++11 多线程高性能库ConcurrentQueue（多生产者、多消费者）

---

#1.简介
ConcurrentQueue是一个header only多线程库

A fast multi-producer, multi-consumer lock-free concurrent queue for C++11

Github地址：https://github.com/cameron314/concurrentqueue

---

#2.使用方法
>1.将以下2个文件复制到项目目录即可使用
concurrentqueue.h
blockingconcurrentqueue.h

>2.代码使用

#include "concurrentqueue.h"
#include "blockingconcurrentqueue.h"

using namespace std;
using namespace moodycamel;
1
2
3
4
5

---

---

#. 全部Api，伪代码

# Allocates more memory if necessary
enqueue(item) : bool
enqueue(prod_token, item) : bool
enqueue_bulk(item_first, count) : bool
enqueue_bulk(prod_token, item_first, count) : bool

# Fails if not enough memory to enqueue
try_enqueue(item) : bool
try_enqueue(prod_token, item) : bool
try_enqueue_bulk(item_first, count) : bool
try_enqueue_bulk(prod_token, item_first, count) : bool

# Attempts to dequeue from the queue (never allocates)
try_dequeue(item&) : bool
try_dequeue(cons_token, item&) : bool
try_dequeue_bulk(item_first, max) : size_t
try_dequeue_bulk(cons_token, item_first, max) : size_t

# If you happen to know which producer you want to dequeue from
try_dequeue_from_producer(prod_token, item&) : bool
try_dequeue_bulk_from_producer(prod_token, item_first, max) : size_t

# A not-necessarily-accurate count of the total number of elements
size_approx() : size_t
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

---

---

#. 基础的使用方法，自动开辟内存空间

ConcurrentQueue<int> q;
q.enqueue(25);

int item;
bool found = q.try_dequeue(item);
assert(found && item == 25);
1
2
3
4
5
6

---

#. 阻塞线程中的使用方法

BlockingConcurrentQueue<int> q;
thread producer([&]() {
    for (int i = 0; i != 100; ++i) {
        this_thread::sleep_for(chrono::milliseconds(i % 10));
        q.enqueue(i);
        cout << "producer:" << to_string(i) << endl;
    }
});

thread consumer([&]() {
    for (int i = 0; i != 100; ++i) {
        int item;
        q.wait_dequeue(item);
        assert(item == i);

        if (q.wait_dequeue_timed(item, chrono::milliseconds(5))) {
            ++i;
            assert(item == i);
            cout << "consumer:" << to_string(i) << endl;
        }
    }
});
producer.join();
consumer.join();

assert(q.size_approx() == 0);
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

---

#. 高效率Token使用

// speed up token
ConcurrentQueue<int> q;
ProducerToken ptok(q);
q.enqueue(ptok, 17);

ConsumerToken ctok(q);
int item;
q.try_dequeue(ctok, item);
assert(item == 17);
1
2
3
4
5
6
7
8
9

---

#. 批处理Bulk使用

ConcurrentQueue<int> q;
int items[] = {11, 22, 33, 44, 55};
q.enqueue_bulk(items, 5);

int result[5];
int count = q.try_dequeue_bulk(result, 5);
for (int i = 0; i != count; ++i) {
    assert(result[i] == items[i]);
}
1
2
3
4
5
6
7
8
9

---