基于无锁队列和c++11的高性能线程池

时间：2021-10-25 21:22:13
基于无锁队列和c++11的高性能线程池
线程使用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.cpp
TEST_SRC_FILES = src/main.cpp
INCLUDE_DIR = src
STD_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 : clean
clean :
    rm -f test_workthreadpool main.o libworkthreadpool.so

相关文章