摘要：
在上述任何业务场景中，我们都需要一个管理类，其职责是管理一堆线程和要执行的相同类型的任务集合。线程将等待执行馈送给它的任务。当任务集大于线程集的数量时，任务将在队列中等待；当线程集合的数量大于任务集合的数量时，线程将处于阻塞和等待状态，执行器将相应地处于不饱和状态。
1 前言
一个后台实时处理的业务平台，通常我们会根据数据的输入与输出，依据时间轴进行分解成不同阶段或不同粒度的逻辑任务，而每一个待处理的数据我们称为任务或者消息。任务之间的关系可以分为两类：a 上下游父子关系，b 可以并行运行的兄弟关系。具有上下游关系的任务集合具有逻辑或数据依赖关系，即上游任务执行完后，才能执行下游任务；具有兄弟关系的任务间逻辑上互不影响，可以并行运行。
无论是上面任一情况的业务场景，我们都需要一种管理类，其职责：管理着一堆线程及其待执行的同类型任务集合。线程会等待去执行喂给它的任务，当任务集合大于线程集合的个数时，任务会在队列排队等待；而当线程集合个数大于任务集合时，线程会挂起处于阻塞等待状态，执行器也相应地处于不饱和状态。在jdk里面有现成的管理类ThreadPoolExecutor，那么在c++里面看看类似的实现吧：
2 任务与任务池
2.1任务
无论是消息或业务数据，可以抽象地表达为：
struct data_pair { char *data; int len; }
2.2 任务池
任务的缓存用队列表达：
std::queue<data_pair*> _queue;
2.3 任务提交入口
int CQueueThread::writeData(void *data, int len) { if (data == NULL || len <= 0) { return EXIT_FAILURE; } data_pair *item = new data_pair(); item->data = (char*) malloc(len); assert(item->data != NULL); memcpy(item->data, data, len); item->len = len; _mutex.lock(); _queue.push(item); _mutex.signal(); _mutex.unlock(); return EXIT_SUCCESS; }
3线程池
3.1 线程封装
c++里面类似jdk里面Thread类的封装CThread
{ class CThread { public: /** * 构造函数 */ CThread() { tid = 0; pid = 0; } /** * 起一个线程，开始运行 */ bool start(Runnable *r, void *a) { runnable = r; args = a; return 0 == pthread_create(&tid, NULL, CThread::hook, this); } /** * 等待线程退出 */ void join() { if (tid) { pthread_join(tid, NULL); tid = 0; pid = 0; } } /** * 得到Runnable对象 * * @return Runnable */ Runnable *getRunnable() { return runnable; } /** * 得到回调参数 * * @return args */ void *getArgs() { return args; } /*** * 得到线程的进程ID */ int getpid() { return pid; } /** * 线程的回调函数 * */ static void *hook(void *arg) { CThread *thread = (CThread*) arg; thread->pid = gettid(); if (thread->getRunnable()) { thread->getRunnable()->run(thread, thread->getArgs()); } return (void*) NULL; } private: /** * 得到tid号 */ #ifdef _syscall0 static _syscall0(pid_t,gettid) #else static pid_t gettid() { return static_cast<pid_t>(syscall(__NR_gettid));} #endif private: pthread_t tid; // pthread_self() id int pid; // 线程的进程ID Runnable *runnable; void *args; }; }
3.2 线程池
并行处理的能力有线程池的个数决定，定义如下：
CThread *_thread; int _threadCount; |
4 执行器
4.1 执行启动
int CDefaultRunnable::start() { if (_thread != NULL || _threadCount < 1) { TBSYS_LOG(ERROR, "start failure, _thread: %p, threadCount: %d", _thread, _threadCount); return 0; } _thread = new CThread[_threadCount]; if (NULL == _thread) { TBSYS_LOG(ERROR, "create _thread object failed, threadCount: %d", _threadCount); return 0; } int i = 0; for (; i<_threadCount; i++) { if (!_thread[i].start(this, (void*)((long)i))) { return i; } } return i; }
4.2 执行
执行器包含了具体业务的执行：
void CQueueThread::run(CThread *thread, void *args) { int threadIndex = (int)((long)(args)); _mutex.lock(); while(!_stop) { while(_stop == 0 && _queue.empty()) { _mutex.wait(); } if (_stop) { break; } data_pair *item = _queue.front(); _queue.pop(); _mutex.unlock(); if (item != NULL) { if (_handler) { _handler->handleQueue(item->data, item->len, threadIndex, _args); } if (item->data) { free(item->data); } free(item); } _mutex.lock(); } _mutex.unlock();
5 样例代码
CMyHandler handler; CQueueThread queueThread(3, &handler, NULL); queueThread.start(); char data[1024];
for(int i=1; i<=mWriteCount; i++) { int len = sprintf(data, "data_%05d", i); queueThread.writeData(data, len+1); } queueThread.wait();
参考：