一直对多线程编程这一块很陌生,决定花一点时间整理一下。
os:ubuntu 10.04 c++
1.最基础,进程同时创建5个线程,各自调用同一个函数
1 #include <iostream> 2 #include <pthread.h> //多线程相关操作头文件,可移植众多平台 3 4 using namespace std; 5 6 #define NUM_THREADS 5 //线程数 7 8 void* say_hello( void* args ) 9 { 10 cout << "hello..." << endl; 11 } //函数返回的是函数指针,便于后面作为参数 12 13 int main() 14 { 15 pthread_t tids[NUM_THREADS]; //线程id 16 for( int i = 0; i < NUM_THREADS; ++i ) 17 { 18 int ret = pthread_create( &tids[i], NULL, say_hello, NULL ); //参数:创建的线程id,线程参数,线程运行函数的起始地址,运行函数的参数 19 if( ret != 0 ) //创建线程成功返回0 20 { 21 cout << "pthread_create error:error_code=" << ret << endl; 22 } 23 } 24 pthread_exit( NULL ); //等待各个线程退出后,进程才结束,否则进程强制结束,线程处于未终止的状态 25 }
输入命令:g++ -o muti_thread_test_1 muti_thread_test_1.cpp -lpthread
注意:
1)此为c++程序,故用g++来编译生成可执行文件,并且要调用处理多线程操作相关的静态链接库文件pthread。
2)-lpthread 编译选项到位置可任意,如g++ -lpthread -o muti_thread_test_1 muti_thread_test_1.cpp
3)注意gcc和g++的区别,转到此文:点击打开链接
测试结果:
1 wq@wq-desktop:~/coding/muti_thread$ ./muti_thread_test_1 2 hello...hello... 3 hello... 4 hello... 5 6 hello...
1 wq@wq-desktop:~/coding/muti_thread$ ./muti_thread_test_1 2 hello...hello...hello... 3 4 hello... 5 hello...
可知,两次运行的结果会有差别,这不是多线程的特点吧?这显然没有同步?还有待进一步探索...
多线程的运行是混乱的,混乱就是正常?
2.线程调用到函数在一个类中,那必须将该函数声明为静态函数函数
因为静态成员函数属于静态全局区,线程可以共享这个区域,故可以各自调用。
1 #include <iostream> 2 #include <pthread.h> 3 4 using namespace std; 5 6 #define NUM_THREADS 5 7 8 class Hello 9 { 10 public: 11 static void* say_hello( void* args ) 12 { 13 cout << "hello..." << endl; 14 } 15 }; 16 17 int main() 18 { 19 pthread_t tids[NUM_THREADS]; 20 for( int i = 0; i < NUM_THREADS; ++i ) 21 { 22 int ret = pthread_create( &tids[i], NULL, Hello::say_hello, NULL ); 23 if( ret != 0 ) 24 { 25 cout << "pthread_create error:error_code" << ret << endl; 26 } 27 } 28 pthread_exit( NULL ); 29 }
测试结果:
1 wq@wq-desktop:~/coding/muti_thread$ ./muti_thread_test_2 2 hello... 3 hello... 4 hello... 5 hello... 6 hello...
1 wq@wq-desktop:~/coding/muti_thread$ ./muti_thread_test_2 2 hello...hello...hello... 3 4 5 hello... 6 hello...
3.如何在线程调用函数时传入参数呢?
先看下面修改的代码,传入线程编号作为参数:
1 #include <iostream> 2 #include <pthread.h> //多线程相关操作头文件,可移植众多平台 3 4 using namespace std; 5 6 #define NUM_THREADS 5 //线程数 7 8 void* say_hello( void* args ) 9 { 10 int i = *( (int*)args ); //对传入的参数进行强制类型转换,由无类型指针转变为整形指针,再用*读取其指向到内容 11 cout << "hello in " << i << endl; 12 } //函数返回的是函数指针,便于后面作为参数 13 14 int main() 15 { 16 pthread_t tids[NUM_THREADS]; //线程id 17 cout << "hello in main.." << endl; 18 for( int i = 0; i < NUM_THREADS; ++i ) 19 { 20 int ret = pthread_create( &tids[i], NULL, say_hello, (void*)&i ); //传入到参数必须强转为void*类型,即无类型指针,&i表示取i的地址,即指向i的指针 21 cout << "Current pthread id = " << tids[i] << endl; //用tids数组打印创建的进程id信息 22 if( ret != 0 ) //创建线程成功返回0 23 { 24 cout << "pthread_create error:error_code=" << ret << endl; 25 } 26 } 27 pthread_exit( NULL ); //等待各个线程退出后,进程才结束,否则进程强制结束,线程处于未终止的状态 28 }
测试结果:
显然不是想要的结果,调用顺序很乱,这是为什么呢?
这是因为多线程到缘故,主进程还没开始对i赋值,线程已经开始跑了...?
修改代码如下:
1 #include <iostream> 2 #include <pthread.h> //多线程相关操作头文件,可移植众多平台 3 4 using namespace std; 5 6 #define NUM_THREADS 5 //线程数 7 8 void* say_hello( void* args ) 9 { 10 cout << "hello in thread " << *( (int *)args ) << endl; 11 } //函数返回的是函数指针,便于后面作为参数 12 13 int main() 14 { 15 pthread_t tids[NUM_THREADS]; //线程id 16 int indexes[NUM_THREADS]; //用来保存i的值避免被修改 17 18 for( int i = 0; i < NUM_THREADS; ++i ) 19 { 20 indexes[i] = i; 21 int ret = pthread_create( &tids[i], NULL, say_hello, (void*)&(indexes[i]) ); 22 if( ret != 0 ) //创建线程成功返回0 23 { 24 cout << "pthread_create error:error_code=" << ret << endl; 25 } 26 } 27 for( int i = 0; i < NUM_THREADS; ++i ) 28 pthread_join( tids[i], NULL ); //pthread_join用来等待一个线程的结束,是一个线程阻塞的函数 29 }
测试结果:
1 wq@wq-desktop:~/coding/muti_thread$ ./muti_thread_test_3 2 hello in thread hello in thread hello in thread hello in thread hello in thread 30124
这是正常的吗?感觉还是有问题...待续
代码中如果没有pthread_join主线程会很快结束从而使整个进程结束,从而使创建的线程没有机会开始执行就结束了。加入pthread_join后,主线程会一直等待直到等待的线程结束自己才结束,使创建的线程有机会执行。
4.线程创建时属性参数的设置pthread_attr_t及join功能的使用
线程的属性由结构体pthread_attr_t进行管理。
typedef struct
{
int detachstate; 线程的分离状态
int schedpolicy; 线程调度策略
struct sched_param schedparam; 线程的调度参数
int inheritsched; 线程的继承性
int scope; 线程的作用域
size_t guardsize; 线程栈末尾的警戒缓冲区大小
int stackaddr_set; void * stackaddr; 线程栈的位置
size_t stacksize; 线程栈的大小
}pthread_attr_t;
1 #include <iostream> 2 #include <pthread.h> 3 4 using namespace std; 5 6 #define NUM_THREADS 5 7 8 void* say_hello( void* args ) 9 { 10 cout << "hello in thread " << *(( int * )args) << endl; 11 int status = 10 + *(( int * )args); //线程退出时添加退出的信息,status供主程序提取该线程的结束信息 12 pthread_exit( ( void* )status ); 13 } 14 15 int main() 16 { 17 pthread_t tids[NUM_THREADS]; 18 int indexes[NUM_THREADS]; 19 20 pthread_attr_t attr; //线程属性结构体,创建线程时加入的参数 21 pthread_attr_init( &attr ); //初始化 22 pthread_attr_setdetachstate( &attr, PTHREAD_CREATE_JOINABLE ); //是设置你想要指定线程属性参数,这个参数表明这个线程是可以join连接的,join功能表示主程序可以等线程结束后再去做某事,实现了主程序和线程同步功能 23 for( int i = 0; i < NUM_THREADS; ++i ) 24 { 25 indexes[i] = i; 26 int ret = pthread_create( &tids[i], &attr, say_hello, ( void* )&( indexes[i] ) ); 27 if( ret != 0 ) 28 { 29 cout << "pthread_create error:error_code=" << ret << endl; 30 } 31 } 32 pthread_attr_destroy( &attr ); //释放内存 33 void *status; 34 for( int i = 0; i < NUM_THREADS; ++i ) 35 { 36 int ret = pthread_join( tids[i], &status ); //主程序join每个线程后取得每个线程的退出信息status 37 if( ret != 0 ) 38 { 39 cout << "pthread_join error:error_code=" << ret << endl; 40 } 41 else 42 { 43 cout << "pthread_join get status:" << (long)status << endl; 44 } 45 } 46 }
测试结果:
1 wq@wq-desktop:~/coding/muti_thread$ ./muti_thread_test_4 2 hello in thread hello in thread hello in thread hello in thread 0hello in thread 321 3 4 5 6 4 7 pthread_join get status:10 8 pthread_join get status:11 9 pthread_join get status:12 10 pthread_join get status:13 11 pthread_join get status:14
5.互斥锁的实现
互斥锁是实现线程同步的一种机制,只要在临界区前后对资源加锁就能阻塞其他进程的访问。
1 #include <iostream> 2 #include <pthread.h> 3 4 using namespace std; 5 6 #define NUM_THREADS 5 7 8 int sum = 0; //定义全局变量,让所有线程同时写,这样就需要锁机制 9 pthread_mutex_t sum_mutex; //互斥锁 10 11 void* say_hello( void* args ) 12 { 13 cout << "hello in thread " << *(( int * )args) << endl; 14 pthread_mutex_lock( &sum_mutex ); //先加锁,再修改sum的值,锁被占用就阻塞,直到拿到锁再修改sum; 15 cout << "before sum is " << sum << " in thread " << *( ( int* )args ) << endl; 16 sum += *( ( int* )args ); 17 cout << "after sum is " << sum << " in thread " << *( ( int* )args ) << endl; 18 pthread_mutex_unlock( &sum_mutex ); //释放锁,供其他线程使用 19 pthread_exit( 0 ); 20 } 21 22 int main() 23 { 24 pthread_t tids[NUM_THREADS]; 25 int indexes[NUM_THREADS]; 26 27 pthread_attr_t attr; //线程属性结构体,创建线程时加入的参数 28 pthread_attr_init( &attr ); //初始化 29 pthread_attr_setdetachstate( &attr, PTHREAD_CREATE_JOINABLE ); //是设置你想要指定线程属性参数,这个参数表明这个线程是可以join连接的,join功能表示主程序可以等线程结束后再去做某事,实现了主程序和线程同步功能 30 pthread_mutex_init( &sum_mutex, NULL ); //对锁进行初始化 31 32 for( int i = 0; i < NUM_THREADS; ++i ) 33 { 34 indexes[i] = i; 35 int ret = pthread_create( &tids[i], &attr, say_hello, ( void* )&( indexes[i] ) ); //5个进程同时去修改sum 36 if( ret != 0 ) 37 { 38 cout << "pthread_create error:error_code=" << ret << endl; 39 } 40 } 41 pthread_attr_destroy( &attr ); //释放内存 42 void *status; 43 for( int i = 0; i < NUM_THREADS; ++i ) 44 { 45 int ret = pthread_join( tids[i], &status ); //主程序join每个线程后取得每个线程的退出信息status 46 if( ret != 0 ) 47 { 48 cout << "pthread_join error:error_code=" << ret << endl; 49 } 50 } 51 cout << "finally sum is " << sum << endl; 52 pthread_mutex_destroy( &sum_mutex ); //注销锁 53 }
测试结果:
可知,sum的访问和修改顺序是正常的,这就达到了多线程的目的了,但是线程的运行顺序是混乱的,混乱就是正常?
6.信号量的实现
信号量是线程同步的另一种实现机制,信号量的操作有signal和wait,本例子采用条件信号变量pthread_cond_t tasks_cond;
信号量的实现也要给予锁机制。
1 #include <iostream> 2 #include <pthread.h> 3 #include <stdio.h> 4 5 using namespace std; 6 7 #define BOUNDARY 5 8 9 int tasks = 10; 10 pthread_mutex_t tasks_mutex; //互斥锁 11 pthread_cond_t tasks_cond; //条件信号变量,处理两个线程间的条件关系,当task>5,hello2处理,反之hello1处理,直到task减为0 12 13 void* say_hello2( void* args ) 14 { 15 pthread_t pid = pthread_self(); //获取当前线程id 16 cout << "[" << pid << "] hello in thread " << *( ( int* )args ) << endl; 17 18 bool is_signaled = false; //sign 19 while(1) 20 { 21 pthread_mutex_lock( &tasks_mutex ); //加锁 22 if( tasks > BOUNDARY ) 23 { 24 cout << "[" << pid << "] take task: " << tasks << " in thread " << *( (int*)args ) << endl; 25 --tasks; //modify 26 } 27 else if( !is_signaled ) 28 { 29 cout << "[" << pid << "] pthread_cond_signal in thread " << *( ( int* )args ) << endl; 30 pthread_cond_signal( &tasks_cond ); //signal:向hello1发送信号,表明已经>5 31 is_signaled = true; //表明信号已发送,退出此线程 32 } 33 pthread_mutex_unlock( &tasks_mutex ); //解锁 34 if( tasks == 0 ) 35 break; 36 } 37 } 38 39 void* say_hello1( void* args ) 40 { 41 pthread_t pid = pthread_self(); //获取当前线程id 42 cout << "[" << pid << "] hello in thread " << *( ( int* )args ) << endl; 43 44 while(1) 45 { 46 pthread_mutex_lock( &tasks_mutex ); //加锁 47 if( tasks > BOUNDARY ) 48 { 49 cout << "[" << pid << "] pthread_cond_signal in thread " << *( ( int* )args ) << endl; 50 pthread_cond_wait( &tasks_cond, &tasks_mutex ); //wait:等待信号量生效,接收到信号,向hello2发出信号,跳出wait,执行后续 51 } 52 else 53 { 54 cout << "[" << pid << "] take task: " << tasks << " in thread " << *( (int*)args ) << endl; 55 --tasks; 56 } 57 pthread_mutex_unlock( &tasks_mutex ); //解锁 58 if( tasks == 0 ) 59 break; 60 } 61 } 62 63 64 int main() 65 { 66 pthread_attr_t attr; //线程属性结构体,创建线程时加入的参数 67 pthread_attr_init( &attr ); //初始化 68 pthread_attr_setdetachstate( &attr, PTHREAD_CREATE_JOINABLE ); //是设置你想要指定线程属性参数,这个参数表明这个线程是可以join连接的,join功能表示主程序可以等线程结束后再去做某事,实现了主程序和线程同步功能 69 pthread_cond_init( &tasks_cond, NULL ); //初始化条件信号量 70 pthread_mutex_init( &tasks_mutex, NULL ); //初始化互斥量 71 pthread_t tid1, tid2; //保存两个线程id 72 int index1 = 1; 73 int ret = pthread_create( &tid1, &attr, say_hello1, ( void* )&index1 ); 74 if( ret != 0 ) 75 { 76 cout << "pthread_create error:error_code=" << ret << endl; 77 } 78 int index2 = 2; 79 ret = pthread_create( &tid2, &attr, say_hello2, ( void* )&index2 ); 80 if( ret != 0 ) 81 { 82 cout << "pthread_create error:error_code=" << ret << endl; 83 } 84 pthread_join( tid1, NULL ); //连接两个线程 85 pthread_join( tid2, NULL ); 86 87 pthread_attr_destroy( &attr ); //释放内存 88 pthread_mutex_destroy( &tasks_mutex ); //注销锁 89 pthread_cond_destroy( &tasks_cond ); //正常退出 90 }
测试结果:
先在线程2中执行say_hello2,再跳转到线程1中执行say_hello1,直到tasks减到0为止。
1 wq@wq-desktop:~/coding/muti_thread$ ./muti_thread_test_6 2 [3069823856] hello in thread 2 3 [3078216560] hello in thread 1[3069823856] take task: 10 in thread 2 4 5 [3069823856] take task: 9 in thread 2 6 [3069823856] take task: 8 in thread 2 7 [3069823856] take task: 7 in thread 2 8 [3069823856] take task: 6 in thread 2 9 [3069823856] pthread_cond_signal in thread 2 10 [3078216560] take task: 5 in thread 1 11 [3078216560] take task: 4 in thread 1 12 [3078216560] take task: 3 in thread 1 13 [3078216560] take task: 2 in thread 1 14 [3078216560] take task: 1 in thread 1
到此,对多线程编程有了一个初步的了解,当然还有其他实现线程同步的机制,有待进一步探索。
C++ 多线程编程总结
在开发C++程序时,一般在吞吐量、并发、实时性上有较高的要求。设计C++程序时,总结起来可以从如下几点提高效率:
- l 并发
- l 异步
- l 缓存
下面将我平常工作中遇到一些问题例举一二,其设计思想无非以上三点。
1任务队列
1.1 以生产者-消费者模型设计任务队列
生产者-消费者模型是人们非常熟悉的模型,比如在某个服务器程序中,当User数据被逻辑模块修改后,就产生一个更新数据库的任务(produce),投递给IO模块任务队列,IO模块从任务队列中取出任务执行sql操作(consume)。
设计通用的任务队列,示例代码如下:
详细实现可参见:
http://ffown.googlecode.com/svn/trunk/fflib/include/detail/task_queue_impl.h
1 void task_queue_t::produce(const task_t& task_) { 2 lock_guard_t lock(m_mutex); 3 if (m_tasklist->empty()){//! 条件满足唤醒等待线程 4 m_cond.signal(); 5 } 6 m_tasklist->push_back(task_); 7 } 8 int task_queue_t::comsume(task_t& task_){ 9 lock_guard_t lock(m_mutex); 10 while (m_tasklist->empty())//! 当没有作业时,就等待直到条件满足被唤醒{ 11 if (false == m_flag){ 12 return -1; 13 } 14 m_cond.wait(); 15 } 16 task_ = m_tasklist->front(); 17 m_tasklist->pop_front(); 18 return 0; 19 }
1.2 任务队列使用技巧
1.2.1 IO 与 逻辑分离
比如网络游戏服务器程序中,网络模块收到消息包,投递给逻辑层后立即返回,继续接受下一个消息包。逻辑线程在一个没有io操作的环境下运行,以保障实时性。示例:
1 void handle_xx_msg(long uid, const xx_msg_t& msg){ 2 logic_task_queue->post(boost::bind(&servie_t::proces, uid, msg)); 3 }
注意,此模式下为单任务队列,每个任务队列单线程。
1.2.2 并行流水线
上面的只是完成了io 和 cpu运算的并行,而cpu中逻辑操作是串行的。在某些场合,cpu逻辑运算部分也可实现并行,如游戏中用户A种菜和B种菜两种操作是完全可以并行的,因为两个操作没有共享数据。最简单的方式是A、B相关的操作被分配到不同的任务队列中。示例如下:
1 void handle_xx_msg(long uid, const xx_msg_t& msg) { 2 logic_task_queue_array[uid % sizeof(logic_task_queue_array)]->post( 3 boost::bind(&servie_t::proces, uid, msg)); 4 }
注意,此模式下为多任务队列,每个任务队列单线程。
1.2.3 连接池与异步回调
比如逻辑Service模块需要数据库模块异步载入用户数据,并做后续处理计算。而数据库模块拥有一个固定连接数的连接池,当执行SQL的任务到来时,选择一个空闲的连接,执行SQL,并把SQL 通过回调函数传递给逻辑层。其步骤如下:
- n 预先分配好线程池,每个线程创建一个连接到数据库的连接
- n 为数据库模块创建一个任务队列,所有线程都是这个任务队列的消费者
- n 逻辑层想数据库模块投递sql执行任务,同时传递一个回调函数来接受sql执行结果
示例如下:
1 void db_t:load(long uid_, boost::function<void (user_data_t&) func_){ 2 //! sql execute, construct user_data_t user 3 func_(user) 4 } 5 void process_user_data_loaded(user_data_t&){ 6 //! todo something 7 } 8 db_task_queue->post(boost::bind(&db_t:load, uid, func));
注意,此模式下为单任务队列,每个任务队列多线程。
2. 日志
本文主要讲C++多线程编程,日志系统不是为了提高程序效率,但是在程序调试、运行期排错上,日志是无可替代的工具,相信开发后台程序的朋友都会使用日志。常见的日志使用方式有如下几种:
- n 流式,如logstream << "start servie time[%d]" << time(0) << " app name[%s]" << app_string.c_str() << endl;
- n Printf 格式如:logtrace(LOG_MODULE, "start servie time[%d] app name[%s]", time(0), app_string.c_str());
二者各有优缺点,流式是线程安全的,printf格式格式化字符串会更直接,但缺点是线程不安全,如果把app_string.c_str() 换成app_string (std::string),编译被通过,但是运行期会crash(如果运气好每次都crash,运气不好偶尔会crash)。我个人钟爱printf风格,可以做如下改进:
- l 增加线程安全,利用C++模板的traits机制,可以实现线程安全。示例:
1 template<typename ARG1> 2 void logtrace(const char* module, const char* fmt, ARG1 arg1){ 3 boost::format s(fmt); 4 f % arg1; 5 }
这样,除了标准类型+std::string 传入其他类型将编译不能通过。这里只列举了一个参数的例子,可以重载该版本支持更多参数,如果你愿意,可以支持9个参数或更多。
- l 为日志增加颜色,在printf中加入控制字符,可以再屏幕终端上显示颜色,Linux下示例:printf("\033[32;49;1m [DONE] \033[39;49;0m")
更多颜色方案参见:
http://hi.baidu.com/jiemnij/blog/item/d95df8c28ac2815cb219a80e.html
- l 每个线程启动时,都应该用日志打印该线程负责什么功能。这样,程序跑起来的时候通过top –H – p pid 可以得知那个功能使用cpu的多少。实际上,我的每行日志都会打印线程id,此线程id非pthread_id,而其实是线程对应的系统分配的进程id号。
3. 性能监控
尽管已经有很多工具可以分析c++程序运行性能,但是其大部分还是运行在程序debug阶段。我们需要一种手段在debug和release阶段都能监控程序,一方面得知程序瓶颈之所在,一方面尽早发现哪些组件在运行期出现了异常。
通常都是使用gettimeofday 来计算某个函数开销,可以精确到微妙。可以利用C++的确定性析构,非常方便的实现获取函数开销的小工具,示例如下
1 struct profiler{ 2 profiler(const char* func_name){ 3 gettimeofday(&tv, NULL); 4 m_func_name=func_name; 5 } 6 ~profiler(){ 7 struct timeval tv2; 8 gettimeofday(&tv2, NULL); 9 long cost = (tv.tv_sec - tv.tv_sec) * 1000000 + (tv.tv_usec - tv.tv_usec); 10 //! post to some manager 11 } 12 struct timeval tv; 13 const char * m_func_name; 14 }; 15 #define PROFILER() profiler ____profiler_instance##__LINE__(__FUNCTION__)
Cost 应该被投递到性能统计管理器中,该管理器定时讲性能统计数据输出到文件中。
4 Lambda 编程
使用foreach 代替迭代器
很多编程语言已经内建了foreach,但是c++还没有。所以建议自己在需要遍历容器的地方编写foreach函数。习惯函数式编程的人应该会非常钟情使用foreach,使用foreach的好处多多少少有些,如:
http://www.cnblogs.com/chsword/archive/2007/09/28/910011.html
但主要是编程哲学上层面的。
示例:
1 void user_mgr_t::foreach(boost::function<void (user_t&)> func_){ 2 for (iterator it = m_users.begin(); it != m_users.end() ++it){ 3 func_(it->second); 4 } 5 }
比如要实现dump 接口,不需要重写关于迭代器的代码
1 void user_mgr_t:dump(){ 2 struct lambda { 3 static void print(user_t& user){ 4 //! print(tostring(user); 5 } 6 }; 7 this->foreach(lambda::print); 8 }
实际上,上面的代码变通的生成了匿名函数,如果是c++ 11 标准的编译器,本可以写的更简洁一些:
this->foreach([](user_t& user) {} );
但是我大部分时间编写的程序都要运行在centos 上,你知道吗它的gcc版本是gcc 4.1.2, 所以大部分时间我都是用变通的方式使用lambda函数。
Lambda 函数结合任务队列实现异步
常见的使用任务队列实现异步的代码如下:
1 void service_t:async_update_user(long uid){ 2 task_queue->post(boost::bind(&service_t:sync_update_user_impl, this, uid)); 3 } 4 void service_t:sync_update_user_impl(long uid){ 5 user_t& user = get_user(uid); 6 user.update() 7 }
这样做的缺点是,一个接口要响应的写两遍函数,如果一个函数的参数变了,那么另一个参数也要跟着改动。并且代码也不是很美观。使用lambda可以让异步看起来更直观,仿佛就是在接口函数中立刻完成一样。示例代码:
1 void service_t:async_update_user(long uid){ 2 struct lambda { 3 static void update_user_impl(service_t* servie, long uid){ 4 user_t& user = servie->get_user(uid); 5 user.update(); 6 } 7 }; 8 task_queue->post(boost::bind(&lambda:update_user_impl, this, uid)); 9 }
这样当要改动该接口时,直接在该接口内修改代码,非常直观。
5. 奇技淫巧
利用shared_ptr 实现map/reduce
Map/reduce的语义是先将任务划分为多个任务,投递到多个worker中并发执行,其产生的结果经reduce汇总后生成最终的结果。Shared_ptr的语义是什么呢?当最后一个shared_ptr析构时,将会调用托管对象的析构函数。语义和map/reduce过程非常相近。我们只需自己实现讲请求划分多个任务即可。示例过程如下:
- l 定义请求托管对象,加入我们需要在10个文件中搜索“oh nice”字符串出现的次数,定义托管结构体如下:
1 struct reducer{ 2 void set_result(int index, long result) { 3 m_result[index] = result; 4 } 5 ~reducer(){ 6 long total = 0; 7 for (int i = 0; i < sizeof(m_result); ++i){ 8 total += m_result[i]; 9 } 10 //! post total to somewhere 11 } 12 long m_result[10]; 13 };
- l 定义执行任务的 worker
1 void worker_t:exe(int index_, shared_ptr<reducer> ret) { 2 ret->set_result(index, 100); 3 }
- l 将任务分割后,投递给不同的worker
-
1 shared_ptr<reducer> ret(new reducer()); 2 for (int i = 0; i < 10; ++i) 3 { 4 task_queue[i]->post(boost::bind(&worker_t:exe, i, ret)); 5 }