基于无锁队列和c++11的高性能线程池
线程使用c++11库
和线程池之间的消息通讯使用一个简单的无锁消息队列
适用于linux平台，gcc 4.6以上
线程使用c++11库
和线程池之间的消息通讯使用一个简单的无锁消息队列
适用于linux平台，gcc 4.6以上
标签： <无>
代码片段(6)[全屏查看所有代码]
1. [代码]lckfree.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
|
// lckfree.h// Implementation of lock free queue using CAS operations// for simple multi-threading use cases like:// 1. multiple worker to process incoming messages// 2. async processing using a thread pool// 3. simple tcp server deal with async requests// Author: typhoon_1986@163.com// Refrence: http://coolshell.cn/articles/8239.html#ifndef __LCKFREE_H__#define __LCKFREE_H__#include <string>using namespace std;
namespace bfd {
struct LinkNode {
string data;
LinkNode* next;
};typedef struct LinkNode LinkNode;
class LckFreeQueue {
public:
LckFreeQueue();
~LckFreeQueue();
int push(const string &msg);
string pop(); // non-block pop method
// string bpop(); // block pop method bool empty();
private:
LinkNode * head_;
LinkNode * tail_;
bool empty_;
unsigned int length_;
};} // namespace bfd
#endif |
2. [代码]lckfree.cpp
|
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
|
#include <lckfree.h>namespace bfd {
LckFreeQueue::LckFreeQueue(): head_(NULL), tail_(NULL), empty_(true), length_(0) {
head_ = new LinkNode;
head_->next = NULL;
tail_ = head_;
}LckFreeQueue::~LckFreeQueue() { LinkNode *p = head_;
if (p) {
LinkNode *q = p->next;
delete p;
p = q;
}
}int LckFreeQueue::push(const string &msg) {
LinkNode * q = new LinkNode;
q->data = msg;
q->next = NULL;
LinkNode * p = tail_;
LinkNode * oldp = p;
do {
while (p->next != NULL)
p = p->next;
} while( __sync_bool_compare_and_swap(&(p->next), NULL, q) != true); //如果没有把结点链在尾上，再试
__sync_bool_compare_and_swap(&tail_, oldp, q); //置尾结点
return 0;
}string LckFreeQueue::pop() { LinkNode * p;
do{
p = head_;
if (p->next == NULL){
return "";
}
} while( __sync_bool_compare_and_swap(&head_, p, p->next) != true );
return p->next->data;
}bool LckFreeQueue::empty() {
return empty_;
}} |
3. [代码]workthreadpool.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
|
// workthreadpool.h// 一个用于将消息分发给多个进程，并使用多个进程处理的库，工作进程并不返回数据#ifndef __WORK_THREAD_POOL__#define __WORK_THREAD_POOL__#include <stdio.h>#include <thread>#include <queue>#include <string>#include <vector>#include "lckfree.h"using namespace std;
namespace bfd {
class WorkThreadPool {
public:
WorkThreadPool(int size);
virtual ~WorkThreadPool();
// 需要子类继承并实现的函数，每个线程实际执行的内容
virtual void Init() {};
virtual void Finish() {};
virtual void Handle(const string &msg)=0;
// 将消息放入处理队列, 消息只支持string类型
int SendMessage(const string &msg);
int Start();
int Stop();
private:
void Worker();
int size_;
LckFreeQueue msg_queue_; // 线程池的协作基于这个无锁队列
vector<thread> thread_pool_;
};} // namespace
#endif |
4. [代码]workthreadpool.cpp
|
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
|
#include "workthreadpool.h"#include <sstream>#include <unistd.h>namespace bfd {
WorkThreadPool::WorkThreadPool(int size) {
if (size <= 0) { // 最小也需要有1个线程
size_ = 1;
} else {
size_ = size;
}
}WorkThreadPool::~WorkThreadPool() {}int WorkThreadPool::SendMessage(const string &msg) {
msg_queue_.push(msg);
return 0;
}void WorkThreadPool::Worker() {
unsigned int msg_count = 0;
while (1) {
string msg = msg_queue_.pop();
if (msg.empty()) {
printf("no msg got, sleep for 0.1 sec\n");
usleep(100000); // 0.1 sec
continue;
}
if (msg == "__exit__") {
stringstream ss;
ss << "exit worker: " << std::this_thread::get_id() << ", processed: " << msg_count << "..";
printf("%s\n", ss.str().c_str());
return;
}
Handle(msg);
msg_count++;
if (msg_count % 1000 == 0) {
printf("every 1000 msg count\n");
}
}
}int WorkThreadPool::Start() {
for (int i=0; i < size_; i++) {
thread_pool_.push_back( thread(&WorkThreadPool::Worker, this) );
}
return 0;
}int WorkThreadPool::Stop() {
for (int i=0; i < size_; i++) {
SendMessage("__exit__");
}
for (int i=0; i < size_; i++) {
thread_pool_[i].join();
}
return 0;
}} |
5. [代码]main.cpp
|
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
|
#include "workthreadpool.h"#include <sstream>#include <math.h>class MyThreadPool : public bfd::WorkThreadPool {
public:
MyThreadPool(int size) : bfd::WorkThreadPool(size) {
}
void Handle(const string &msg) {
stringstream ss;
ss << "worker (" << std::this_thread::get_id() << ") got msg: " << msg;
printf("%s\n", ss.str().c_str());
for (int i=0; i<=999999; i++) {
double result = sqrt(sqrt(i) / 93.234);
}
}
};int main() {
printf("start running ....\n");
MyThreadPool pool(5);
pool.Start();
for (int i=0; i<100; i++) {
pool.SendMessage("msg info ----------");
}
pool.Stop();
return 0;
} |
6. [代码]Makefile
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
LIB_SRC_FILES = src/workthreadpool.cpp src/lckfree.cppTEST_SRC_FILES = src/main.cppINCLUDE_DIR = srcSTD_FLAG = -std=c++0x all: main.o libs g++ $(STD_FLAG) -o test_workthreadpool main.o libworkthreadpool.so -lpthread
main.o: $(TEST_SRC_FILES) g++ $(STD_FLAG) -c $(TEST_SRC_FILES) -I$(INCLUDE_DIR)
libs: $(LIB_SRC_FILES) g++ $(STD_FLAG) -o libworkthreadpool.so -fPIC -O2 -shared -Wl,--no-as-needed -Isrc $(LIB_SRC_FILES) -lpthread
.PHONY : cleanclean : rm -f test_workthreadpool main.o libworkthreadpool.so
|