您的位置: 网界网 > 网络学院-技术开发 > 正文

C++多进程并发框架FFLIB之Tutorial

2014年06月19日 20:46:11 | 作者:佚名 | 来源:51CTO | 查看本文手机版

摘要:基本介绍可以看这里:     http://www.cnblogs.com/zhiranok/archive/2012/07/30/fflib_framework.html 其中之所以特意采用了Broker模式,是吸收了MPI和Erlang的思想。 关于MPI:http://www.mcs.anl.gov/research...

标签
C++

基本介绍可以看这里:

    http://www.cnblogs.com/zhiranok/archive/2012/07/30/fflib_framework.html

其中之所以特意采用了Broker模式,是吸收了MPI和Erlang的思想。

关于MPI:http://www.mcs.anl.gov/research/projects/mpi/

关于Erlang:http://www.erlang.org/

FFLIB 目前处于alpha阶段,一些有用的功能还需继续添加。但是FFLIB一开始就是为了解决实际问题而生。Broker 即可以以独立进程运行,也可以集成到某个特定的进程中启动。除了这些,FFLIB中使用epoll实现的网络层也极具参考价值。网上有一些关于epoll ET 和 LT的讨论,关于哪种方式更简单,本人的答案是ET。ET模式下epoll 就是一个完全状态机。开发者只需实现FD的read、write、error 三种状态即可。

我进一步挖掘FFLIB的功能。写一篇FFLIB的Tutorial。创建更多的FFLIB使用示例,以此来深入探讨FFLIB的意义。在游戏开发中,或者一些分布式的环境中,有许多大家熟悉的模式。,本文挑选了如下作为FFLIB示例:

Request/Reply

点对点通讯

阻塞通讯

多播通讯

Map/Reduce

Request/Reply

异步的Request/Reply

在FFLIB中所有的消息都是Request和Reply一一对应的,默认情况下工作在异步模式。假设如下场景,Flash连入GatewayServer并发送Login消息包,GatewaServer 解析用户名密码,调用LoginServer 验证。

首先定义msg:

  1. struct user_login_t 
  2.     struct in_t: public msg_i 
  3.     { 
  4.         in_t(): 
  5.             msg_i("user_login_t::in_t"
  6.         {} 
  7.         string encode() 
  8.         { 
  9.             return (init_encoder() << uid << value).get_buff(); 
  10.         } 
  11.         void decode(const string& src_buff_) 
  12.         { 
  13.             init_decoder(src_buff_) >> uid >> value; 
  14.         } 
  15.         long   uid; 
  16.         string value; 
  17.     }; 
  18.     struct out_t: public msg_i 
  19.     { 
  20.         out_t(): 
  21.             msg_i("user_login_t::out_t"
  22.         {} 
  23.         string encode() 
  24.         { 
  25.             return (init_encoder() << value).get_buff(); 
  26.         } 
  27.         void decode(const string& src_buff_) 
  28.         { 
  29.             init_decoder(src_buff_) >> value; 
  30.         } 
  31.         bool value; 
  32.     }; 
  33. }; 

LoginServer中如此定义接口:

  1. class login_server_t 
  2. public
  3.     void verify(user_login_t::in_t& in_msg_, rpc_callcack_t& cb_) 
  4.     { 
  5.         user_login_t::out_t out; 
  6.         out.value = true
  7.         cb_(out); 
  8.     } 
  9. }; 
  10. login_server_t login_server; 
  11. singleton_t::instance().create_service("login_server", 1) 
  12.             .bind_service(&login_server) 
  13.             .reg(&login_server_t::verify); 

在GatewayServer中调用上面接口:

  1. struct lambda_t 
  2.     { 
  3.         static void callback(user_login_t::out_t& msg_, socket_ptr_t socket_) 
  4.         { 
  5.             if (true == msg_.value) 
  6.             { 
  7.                 //! socket_->send_msg("login ok"); 
  8.             } 
  9.             else 
  10.             { 
  11.                 //! socket_->send_msg("login failed"); 
  12.             } 
  13.         } 
  14.     }; 
  15.     user_login_t::in_t in; 
  16.     in.uid  = 520; 
  17.     in.value = "ILoveYou"
  18.     socket_ptr_t flash_socket = NULL;//! TODO 
  19.     singleton_t::instance() 
  20.          .get_service_group("login_server_t"
  21.         ->get_service(1) 
  22.        ->async_call(in, binder_t::callback(&lambda_t::callback, flash_socket)); 

如上所示, async_call 可以通过binder_t模板函数为回调函绑定参数。

同步的Request/Reply

大部分时候我们期望Reply被异步处理,但有时Reply 必须被首先处理后才能触发后续操作,一般这种情况发生在程序初始化之时。假设如下场景,SceneServer启动时必须从SuperServer中获取配置,然后才能执行加载场景数据等后续初始化操作。

首先定义通信的msg:

  1. struct config_t 
  2.     struct in_t: public msg_i 
  3.     { 
  4.         in_t(): 
  5.             msg_i("config_t::in_t"
  6.         {} 
  7.         string encode() 
  8.         { 
  9.             return (init_encoder() << server_type << server_id).get_buff(); 
  10.         } 
  11.         void decode(const string& src_buff_) 
  12.         { 
  13.             init_decoder(src_buff_) >> server_type >> server_id; 
  14.         } 
  15.         int server_type; 
  16.         int server_id; 
  17.     }; 
  18.     struct out_t: public msg_i 
  19.     { 
  20.         out_t(): 
  21.             msg_i("config_t::out_t"
  22.         {} 
  23.         string encode() 
  24.         { 
  25.             return (init_encoder() << value).get_buff(); 
  26.         } 
  27.         void decode(const string& src_buff_) 
  28.         { 
  29.             init_decoder(src_buff_) >> value; 
  30.         } 
  31.         map value; 
  32.     }; 
  33. }; 

如上所示, msg 序列化自动支持map。

SuperServer 中定义返回配置的接口:

  1. super_server_t super_server; 
  2. singleton_t::instance().create_service("super_server", 1) 
  3.     .bind_service(&super_server) 
  4.     .reg(&super_server_t::get_config); 
  5. SceneServer 可以如此实现同步Request/Reply: 
  6. rpc_future_t rpc_future; 
  7. config_t::in_t in; 
  8. in.server_type = 1; 
  9. in.server_id   = 1; 
  10. const config_t::out_t& out = rpc_future.call(  singleton_t::instance().get_service_group("super_server"
  11.         ->get_service(1), in); 
  12. cout << out.value.size() <<"\n"
  13. //std::foreach(out.value.begin(), out.value.end(), fuctor_xx); 

点对点通讯

异步Request/Reply 已经能够解决大部分问题了,但是有时处理Push模式时稍显吃了。我们知道消息推算有Push 和Poll两种方式。了解二者:

    http://blog.sina.com.cn/s/blog_6617106b0100hrm1.html

上面提到的Request/Reply 非常适合poll模式,以上一个获取配置为例,SuperServer由于定义接口的时候只需知道callback,并不知道SceneServer的具体连接。,所以SuperServer不能向SceneServer Push消息。在FFLIB中并没有限定某个节点必须是Client或只能是Service,实际上可以兼有二者的角色。SceneServer 也可以提供接口供SuperServer调用,这就符合了Push的语义。假设如下场景,GatewayServer需要在用户登入时调用通知SessionServer,而某一时刻SessionServer也可能呢通知GatewayServer 强制某用户下线。二者互为client和service。大家必须知道,在FFLIB中实现两个节点的通信只需知道对方的服务名称即可,Broker 在此时实现解耦的作用非常明显,若要增加对其他节点的通信,只需通过服务名称async_call即可。

定义通信的msg:

  1. struct user_online_t 
  2.     struct in_t: public msg_i 
  3.     { 
  4.         in_t(): 
  5.             msg_i("user_online_t::in_t"
  6.         {} 
  7.         string encode() 
  8.         { 
  9.             return (init_encoder() << uid).get_buff(); 
  10.         } 
  11.         void decode(const string& src_buff_) 
  12.         { 
  13.             init_decoder(src_buff_) >> uid; 
  14.         } 
  15.         long uid; 
  16.     }; 
  17.     struct out_t: public msg_i 
  18.     { 
  19.         out_t(): 
  20.             msg_i("user_online_t::out_t"
  21.         {} 
  22.         string encode() 
  23.         { 
  24.             return (init_encoder() << value).get_buff(); 
  25.         } 
  26.         void decode(const string& src_buff_) 
  27.         { 
  28.             init_decoder(src_buff_) >> value; 
  29.         } 
  30.         bool value; 
  31.     }; 
  32. }; 
  33. struct force_user_offline_t 
  34.     struct in_t: public msg_i 
  35.     { 
  36.         in_t(): 
  37.             msg_i("force_user_offline_t::in_t"
  38.         {} 
  39.         string encode() 
  40.         { 
  41.             return (init_encoder() << uid).get_buff(); 
  42.         } 
  43.         void decode(const string& src_buff_) 
  44.         { 
  45.             init_decoder(src_buff_) >> uid; 
  46.         } 
  47.        long uid; 
  48.     }; 
  49.     struct out_t: public msg_i 
  50.     { 
  51.         out_t(): 
  52.             msg_i("force_user_offline_t::out_t"
  53.         {} 
  54.        string encode() 
  55.         { 
  56.             return (init_encoder() << value).get_buff(); 
  57.         } 
  58.         void decode(const string& src_buff_) 
  59.         { 
  60.             init_decoder(src_buff_) >> value; 
  61.         } 
  62.         bool value; 
  63.     }; 
  64. }; 

GatewayServer 通知SessionServer 用户上线,并提供强制用户下线的接口:

  1. class gateway_server_t 
  2. public
  3.     void force_user_offline(force_user_offline_t::in_t& in_msg_, rpc_callcack_t& cb_) 
  4.     { 
  5.         //! close user socket 
  6.         force_user_offline_t::out_t out; 
  7.         out.value = true
  8.         cb_(out); 
  9.     } 
  10. }; 
  11. gateway_server_t gateway_server; 
  12. singleton_t::instance().create_service("gateway_server", 1) 
  13.             .bind_service(&gateway_server) 
  14.             .reg(&gateway_server_t::force_user_offline); 
  15. user_online_t::in_t in; 
  16. in.uid = 520; 
  17. singleton_t::instance() 
  18.     .get_service_group("session_server"
  19.     ->get_service(1) 
  20.     ->async_call(in, callback_TODO); 

SessionServer 提供用户上线接口,可能会调用GatewayServer 的接口强制用户下线。

  1. class session_server_t 
  2. public
  3.     void user_login(user_online_t::in_t& in_msg_, rpc_callcack_t& cb_) 
  4.     { 
  5.         //! close user socket 
  6.         user_online_t::out_t out; 
  7.         out.value = true
  8.         cb_(out); 
  9.     } 
  10. }; 
  11. session_server_t session_server; 
  12. singleton_t::instance().create_service("session_server", 1) 
  13.             .bind_service(&session_server) 
  14.             .reg(&session_server_t::user_login); 
  15. force_user_offline_t::in_t in; 
  16. in.uid = 520; 
  17. singleton_t::instance() 
  18.     .get_service_group("gateway_server"
  19.     ->get_service(1) 
  20.     ->async_call(in, callback_TODO); 

多播通信

和点对点通信一样(+微信关注网络世界),要实现多播,只需要知道目标的服务名称。特别提一点的是,FFLIB中有服务组的概念。比如启动了多个场景服务器SceneServer,除了数据不同,二者接口完全相同,有可能只是相同进程的不同实例。在FFLIB框架中把这些服务归为一个服务组,然后再为每个实例分配索引id。

假设如下场景,SuperServer 中要实现一个GM接口,通知所有SceneServer 重新加载配置。

定义通信的msg:

  1. struct reload_config_t 
  2.  
  3.    struct in_t: public msg_i 
  4.    { 
  5.        in_t(): 
  6.            msg_i("reload_config_t::in_t"
  7.        {} 
  8.        string encode() 
  9.        { 
  10.            return (init_encoder()).get_buff(); 
  11.        } 
  12.        void decode(const string& src_buff_) 
  13.        { 
  14.            init_decoder(src_buff_); 
  15.        } 
  16.   }; 
  17.    struct out_t: public msg_i 
  18.    { 
  19.       out_t(): 
  20.            msg_i("reload_config_t::out_t"
  21.        {} 
  22.         string encode() 
  23.        { 
  24.            return (init_encoder() << value).get_buff(); 
  25.        } 
  26.        void decode(const string& src_buff_) 
  27.        { 
  28.            init_decoder(src_buff_) >> value; 
  29.        } 
  30.        bool value; 
  31.    }; 

SceneServer 提供重新载入配置接口:

  1. class scene_server_t 
  2. public
  3.     void reload_config(reload_config_t::in_t& in_msg_, rpc_callcack_t& cb_) 
  4.     { 
  5.         //! close user socket 
  6.         reload_config_t::out_t out; 
  7.         out.value = true
  8.         cb_(out); 
  9.     } 
  10. }; 
  11. scene_server_t scene_server; 
  12. singleton_t::instance().create_service("scene_server", 1) 
  13.             .bind_service(&scene_server) 
  14.             .reg(&scene_server_t::reload_config);  

在SuperServer 中如此实现多播(跟准确是广播,大同小异):

  1. struct lambda_t 
  2. static void reload_config(rpc_service_t* rs_) 
  3.           reload_config_t::in_t in; 
  4.           rs_->async_call(in, callback_TODO); 
  5. }; 
  6. singleton_t::instance() 
  7.     .get_service_group("scene_server"
  8.     ->foreach(&lambda_t::reload_config); 

Map/Reduce

在游戏中使用Map/reduce 的情形并不多见,本人找到网上最常见的Map/reduce 实例 WordCount。情形如下:有一些文本字符串,统计每个字符出现的次数。

Map操作,将文本分为多个子文本,分发给多个Worker 进程进行统计

Reduce 操作,将多组worker 进程计算的结果汇总

Worker:为文本统计各个字符出现的次数

定义通信消息: 

  1. struct word_count_t 
  2.     struct in_t: public msg_i 
  3.     { 
  4.         in_t(): 
  5.             msg_i("word_count_t::in_t"
  6.         {} 
  7.         string encode() 
  8.         { 
  9.             return (init_encoder() << str).get_buff(); 
  10.         } 
  11.         void decode(const string& src_buff_) 
  12.         { 
  13.             init_decoder(src_buff_) >> str; 
  14.         } 
  15.         string str; 
  16.     }; 
  17.     struct out_t: public msg_i 
  18.     { 
  19.         out_t(): 
  20.             msg_i("word_count_t::out_t"
  21.         {} 
  22.         string encode() 
  23.         { 
  24.             return (init_encoder() << value).get_buff(); 
  25.         } 
  26.         void decode(const string& src_buff_) 
  27.         { 
  28.             init_decoder(src_buff_) >> value; 
  29.         } 
  30.        map<charint> value; 
  31.     }; 
  32.  
  33. }; 

定义woker的接口:

  1. class worker_t 
  2. public
  3.     void word_count(word_count_t::in_t& in_msg_, rpc_callcack_t& cb_) 
  4.     { 
  5.         //! close user socket 
  6.         word_count_t::out_t out; 
  7.         for (size_t i = 0; i < in_msg_.str.size();  i) 
  8.         { 
  9.             map<intint>::iterator it = out.value.find(in_msg_.str[i]); 
  10.             if (it != out.value.end()) 
  11.             { 
  12.                 it->second  = 1; 
  13.             } 
  14.             else 
  15.             { 
  16.                 out.value[in_msg_.str[i]] = 1; 
  17.             } 
  18.         } 
  19.         cb_(out); 
  20.     } 
  21. }; 
  22. worker_t worker; 
  23.    for (int i = 0; i < 5;  i) 
  24.     { 
  25.        singleton_t::instance().create_service("worker", 1) 
  26.             .bind_service(&worker) 
  27.             .reg(&worker_t::word_count); 
  28.     } 

模拟Map/reduce 操作:

  1. struct lambda_t 
  2.     static void reduce(word_count_t::out_t& msg_, map<intint>* result_, size_t* size_) 
  3.     { 
  4.         for (map<intint>::iterator it = msg_.value.begin(); it != msg_.value.end();  it) 
  5.         { 
  6.             map<intint>::iterator it2 = result_->find(it->first); 
  7.             if (it2 != result_->end()) 
  8.             { 
  9.                 it2->second  = it->second; 
  10.             } 
  11.             else 
  12.             { 
  13.                 (*result_)[it->first] = it->second; 
  14.             } 
  15.         } 
  16.         if (-- size_ == 0) 
  17.         { 
  18.             //reduce end!!!!!!!!!!!!!!!! 
  19.             delete result_; 
  20.             delete size_; 
  21.         } 
  22.     } 
  23.     static void do_map(const char** p, size_t size_) 
  24.     { 
  25.         map<intint>* result  = new map<intint>(); 
  26.        size_t*    dest_size   = new size_t(); 
  27.         *dest_size = size_; 
  28.         for (size_t i = 0; i < size_;  i) 
  29.         { 
  30.             word_count_t::in_t in; 
  31.             in.str = p[i]; 
  32.             singleton_t::instance() 
  33.                 .get_service_group("worker"
  34.                 ->get_service(1   i % singleton_t::instance().get_service_group("worker")->size()) 
  35.                ->async_call(in, binder_t::callback(&lambda_t::reduce, result, dest_size)); 
  36.         } 
  37.     } 
  38. }; 
  39. const char* str_vec[] = {"oh nice""oh fuck""oh no""oh dear""oh wonderful""oh bingo"}; 
  40. lambda_t::do_map(str_vec, 6); 

总结:

FFLIB 使进程间通信更容易

source code:  https://ffown.googlecode.com/svn/trunk

原文链接:http://www.cnblogs.com/zhiranok/archive/2012/08/08/fflib_tutorial.html

[责任编辑:孙可 sun_ke@cnw.com.cn]